diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 476a764..7c7dfac 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,6 +37,7 @@ jobs: - run: python -m pip install --upgrade pip - run: python -m pip install -e ".[test]" - run: python -m pytest sparkcache -q + - run: python -m pytest research -q - run: python -m pytest deploy -q package: diff --git a/docs/HEAT_AND_SSD_CONTROL_RESEARCH.md b/docs/HEAT_AND_SSD_CONTROL_RESEARCH.md new file mode 100644 index 0000000..890a9b7 --- /dev/null +++ b/docs/HEAT_AND_SSD_CONTROL_RESEARCH.md @@ -0,0 +1,655 @@ +# Heat-aware admission and SSD write control: design specification + +Status labels follow `CONTRIBUTING.md`. This document is a present-state +specification, not a history. + +| Component | Status | +|---|---| +| 8-bit bounded heat-hit ring (counters, saturation, epoch decay) | implemented (research prototype, `research/heat_ssd_control/`) | +| Verified-restore recomputation-token calculation | implemented (research prototype) | +| Chunk reference ledger, shared-trunk accounting, marginal-byte cost | implemented (research prototype) | +| W-TinyLFU-style shadow cache for admission experiments | implemented (research prototype) | +| Hourly and daily staged-write window reporting | implemented (research prototype) | +| Staged-write versus unique-object byte accounting | implemented (research prototype) | +| SMART/Health log-page parsing and Data Units Written deltas | implemented (research prototype) | +| GPU-free behavior and production-import isolation tests | implemented (`research/heat_ssd_control/test_prototype.py`) | +| Per-rank publication-byte telemetry from the serving connector | implemented; research-ledger ingestion is not connected | +| Wiring heat metadata into the vLLM connector, `ManifestStore`, or any serving path | unsupported | +| Publication or eviction decisions that consume heat metadata | research-only | +| Budget enforcement (rejecting or delaying a publication when over budget) | unsupported | +| Physical (NAND-media) write-amplification attribution | unsupported | +| Acquisition of the SMART/Health log page on a specific platform | unsupported (prototype consumes 512 raw bytes; it does not issue ioctls) | + +This is a research design. No serving behavior described here as +research-only may be assumed by deployment tooling. The implemented interface +of SparkCache remains as specified in `README.md`; in particular `README.md` +(`Operations` section) continues to state, correctly, that hourly write +budgets, daily write budgets, and a physical-write-amplification estimate are +**unsupported** in the deployed connector. + +## 1. Production context + +SparkCache persists rank-local KV context to local NVMe through +`sparkcache.persistent_context_cache.cache_manifest.ManifestStore`: + +- Chunks are immutable, content-addressed files (`chunks/.spcc`, + magic `SPCKV001`, `FORMAT_ABI = 1`). Each chunk carries up to + `CacheIdentity.chunk_tokens` tokens (default 256) split into record + families such as `target_ckv`, `sparse_indexer`, and `mtp_draft_kv`. +- Exact manifests (`manifests//.json`) are the + only visibility edge; prefix alias files (`prefix-aliases/`) and descriptor + segments (`prefix-index/`) reference existing chunks without copying them. + Alias publication is bounded to 64 aliases per exact root with descriptor + segments of at most 16 chunk descriptors. +- Publication is durable and write-visible: `_publish_immutable_batch` writes + each object's complete bytes to a `.writing-` temporary file, fsyncs + the file data, hard-links the temporary into place, and issues one + directory fsync per batch. Identical content at the destination is + verified and **not** overwritten, but the temporary staging bytes are + still written first, so re-committing an unchanged chunk costs its full + encoded size of host writes while retaining zero additional bytes. +- Capacity is governed by `CapacityPolicy` (high watermark + `spark_cache_max_bytes`, low watermark defaulting to 90% of the high + watermark, optional TTL seconds). `ManifestStore.maintain` evicts whole + roots in manifest-recency LRU order and counts shared chunks and alias + segments once. It has no knowledge of request frequency beyond the + best-effort recency touch, which restores may apply at most once per 60 + seconds (`ManifestStore.touch`, `minimum_interval_seconds=60`). +- Scheduler- and worker-side counters in + `sparkcache.spark_context_cache_connector` (for example `restore_hit`, + `load_verified`, `store_skipped_present`, `store_skipped_quorum`, + `restore_skip_backlog`) already observe part of what a heat-aware policy + would need, but they are exported as `SparkCacheStats` only; no retention + or admission decision consumes them. + +Consequences that motivate this design: + +1. The default `snapshot-v1` publication schema republishes a complete + snapshot for a grown conversation. The opt-in `tail-cow-v1` and + `page-tail-cow-v1` identities publish immutable row tails or page deltas + instead. All three paths can stage an object before discovering that its + content-addressed destination already exists, so retained bytes alone do + not describe SSD traffic. +2. There is no frequency signal: a context restored once per second and a + context restored once a day are equally subject to recency-LRU eviction + if their manifests age similarly. +3. The serving connector has no publication-byte ledger or write budget. It + does not correlate publication activity with host-observed NVMe writes. + +## 2. Goals, non-goals, and the heat-isolation contract + +Goals: + +- A bounded, cheap, self-decaying heat signal per stored context. +- A defensible measurement of what heat-aware admission would retain or + reject, produced by a shadow instance that cannot affect serving. +- Accounting that separates logical retained bytes from host-observed + written bytes and from staged (write-path) bytes. +- An operator-facing method to observe SSD wear contribution through the + standard NVMe Data Units Written counter. + +Non-goals: + +- Any change to restore correctness, hit authentication, or verification. +- Any deployment configuration option: no + `--kv-transfer-config` key or environment variable is defined by this + design. + +Heat-isolation contract (binding on any future integration): + +1. Heat metadata is diagnostic-only state that lives outside every + authenticated storage surface. It must never appear in manifests, chunks, + alias or segment payloads, `CacheIdentity`, any digest, or any schema + validated by `_validate_manifest_metadata`. +2. Heat values may influence only *storage admission* (whether a candidate + publication proceeds, and which roots maintenance removes). Heat values + must never influence whether `ManifestStore.lookup` reports a hit, whether + restore accepts bytes, or how `restore()` verifies content. A damaged but + hot entry must be rejected and recomputed exactly like a cold one. +3. Serving must never wait on heat work: evaluation, decay, and budget + computation are off the serving path by construction and must remain so. +4. Losing all heat metadata must be benign. Process restarts, device + replacement, and clear-once (`spark_cache_clear_once`) may destroy heat + state wholesale; correctness is unaffected, and the only consequence is + the same unobserved posture as an empty heat model. + +## 3. Bounded 8-bit hit counters + +### Semantics + +Each stored context is represented by a slot in a fixed-size ring of +unsigned 8-bit saturating counters. The ring is process-local scratch state; +it is never written to disk, never synced, and never shared across ranks. + +- Key: `HeatKey(storage_key, context_digest)` — both fields are 64-character + lowercase SHA-256 hex digests, matching `EntryKey` semantics for exact + manifests (`EntryKey.root_kind = "manifest"`). Prefix-alias entries use + the same pair because an alias hit resolves through the source manifest's + identity; alias-specific hit accounting is not modeled. +- Slot index: little-endian 64-bit BLAKE2b digest of + `` `0x0A` ``, taken modulo the ring capacity. + Capacity must be a power of two, so the modulo is byte masking: with + `digest_size = 8`, take the low `log2(capacity)` bits. +- Increment trigger: one increment per **completed verified restore** of the + pair — the worker counted `load_verified` for that digest — and, in + admission experiments, one increment per accepted publication candidate. + Scheduler-probe manifest matches (`restore_hit` before verification) must + not increment production-consumable counters in any integrated form; + unverified probe counts are advisory. +- Saturation: increments clamp at 255; no wrap. +- Decay: every `decay_window` total increments (default 8192), every nonzero + counter is shifted right by `decay_shift` bits (default 1) in one pass and + the in-flight increment applies after the sweep. This bounds per-key + overestimation from hash collisions to the collision rate and bounds the + memory to one byte per slot. + +### Sized example + +A ring of 131,072 slots is 128 KiB of process memory; every slot has 256 +representable values. With `decay_window = 8192`, a key restored once per decay +window settles at a steady estimate of about 1; a key restored ten times per +window settles near 10 with 8-bit precision until saturation. Keys that +collide in one slot add their estimates (false sharing); the design accepts +this because admission decisions are comparative, not absolute. + +### Rejection and bounded behavior + +- Malformed keys (digest shape), non-power-of-two capacity, or a decay + window below 1 raise `ResearchFormatError` at construction or use. +- The ring has no I/O and no locks: a single-process caller is assumed. +- Any byte-level snapshot round trip is schema-checked; a mismatch raises + `ResearchFormatError` rather than guessing. +- Counters are exact-lossy by design. There is no transition in which a + counter drives a correctness decision, because per the isolation contract + no correctness decision ever reads it. + +### Schemas + +`HitRing.snapshot()` emits JSON with schema +`sparkcache-research-heat-ring/v1`: + +```json +{ + "schema": "sparkcache-research-heat-ring/v1", + "capacity": 131072, + "decay_window": 8192, + "decay_shift": 1, + "increments_since_decay": 4112, + "counts_hex": "<262144 hex characters; one byte per slot, index 0 first>" +} +``` + +`HitRing.from_json(payload)` accepts only this schema and exact-length count +arrays. A snapshot preserves counters but not key identities; slot lookup +after a reload is still deterministic, so estimates for known keys survive +restart. An unobserved key can inherit a nonzero estimate when it collides +with an occupied slot, which is the sketch's stated false-sharing tradeoff. + +## 4. Recomputation tokens avoided + +Every verified restore replaces prefill work over the restored span. For a +restored manifest with `committed_tokens = s` accepted at +`num_computed_tokens = c` already-scheduled prefix tokens, the avoided +recomputation for that request is: + +``` +recompute_tokens_avoided = s - c +``` + +`recomputation_tokens_avoided(s, c)` implements this calculation and rejects +negative, non-integral, or reversed spans. + +Over a wall-clock window `W`, for restores `i` of a context: + +``` +tokens_avoided(W, context) = sum_i (s_i - c_i) +``` + +When `c_i` is unavailable for a historical trace, the design approximates +`c_i = 0` and labels the result `tokens_avoided_upper_bound` — the +approximation overcounts only by prefix overlap with concurrently scheduled +work, which the production path reports when present. + +A storage value comparison per context over a retention horizon `T`: + +``` +value(context) = tokens_avoided(T, context) # benefit side +cost(context) = committed_tokens # publication tokens + x rank_shard_factor + x republish_factor(context growth in T) +``` + +- `rank_shard_factor` accounts for per-rank sharding: under DCP degree `d`, + each rank stores its shard of the span, so the fleet writes approximately + `d` shards' worth of the same logical context; a per-rank record counts + only its shard. +- `republish_factor` is 1 for a stable context and grows with the number of + distinct publications of the same conversation. `snapshot-v1` submits a + complete snapshot; `tail-cow-v1` and `page-tail-cow-v1` submit only the + immutable tail or page-semantic delta plus authenticated metadata. + +The prototype does not compute `value` itself. It provides the per-key +counter that any trace-summation script combines with actual restore timing +records (`sparkcache-restore-timing/v1`, which record the selected span) to +derive it. Deriving it automatically inside the connector is research-only. + +Status: per-request formula is defined; automatic in-connector attribution +**research-only**; any admission decision consuming it **research-only**. + +## 5. Shared-trunk value + +A **trunk** is a chunk-aligned token prefix shared by at least two stored +contexts under the same `storage_key`. Chunks are content-addressed, so two +contexts share trunk bytes exactly when the same chunk digests appear in +both manifests (or in an alias's descriptor chain rooted at the same +manifest). The prefix-alias machinery already produces this shape: one exact +manifest plus up to 64 alias roots whose descriptor chains reference the +identical chunk objects. + +For a stored context `C` with chunk list `chunks(C)` (ordered by logical +range) and the ledger's reference count `refs(d)` for chunk digest `d`: + +``` +shared_chunks(C) = { d in chunks(C) : refs(d) >= 2 } +trunk_tokens(C) = sum(token_count(d) for d in shared_chunks(C)) +``` + +- The count of shared chunks equals the shared prefix length in the actual + publication pattern (contexts only append), but the ledger does not + enforce consecutiveness; it reports the reference-count fact and leaves + prefix interpretation to the caller. +- The value of the trunk to admission/eviction pressure: a context whose + chunks are widely referenced contributes low *marginal* bytes (section 6) + and high *avoided tokens per retained byte* — evicting it costs + `trunk_tokens` of recomputation for every referencing root, while evicting + a fully exclusive context costs one root's span. Manifest-recency LRU has + no such notion; `ROADMAP.md` ("Trunk-aware eviction") describes the + same gap and names alias reference counts as the prerequisite. This design + supplies those counts as the ledger. + +The prototype models all of this in `ChunkLedger`: it records exact per-chunk +token spans and byte sizes, decrements references on removal, and reports +`ContextHeatReport(shared_chunk_count, shared_tokens, marginal_bytes, +retained_shared_bytes, chunk_count)` per context. + +## 6. Exclusive physical-byte cost + +The **marginal byte cost** of a stored root is the number of filesystem +bytes that would be reclaimed if that root were deleted and the orphan +collector ran — the quantity `MaintenanceReport.bytes_reclaimed` measures +after the fact and this design predicts before an admission decision: + +``` +marginal_bytes(C) = sum(|b_d| : d in chunks(C), refs(d) == 1) + + manifest_bytes(C) + + sum(|b_seg| : seg in segments(C), seg_refs(seg) == 1) +retained_shared_bytes(C) = sum(|b_d| : d in chunks(C), refs(d) >= 2) +``` + +Chunk byte counts come from chunk descriptors (`descriptor["bytes"]`, the +exact encoded length). Two cost layers sit on top of encoded bytes: + +- **Allocation rounding.** Files occupy whole allocation blocks. + `CommitReceipt.allocated_bytes_upper_bound` already computes the upper + bound `sum(ceil(size / 4096) * 4096)` across the manifest and every chunk; + maintenance instead measures `st_blocks * 512`. Small metadata appearing + large under 4 KiB rounding is one reason small manifests and alias files + cost more physical space than their encoded length suggests. +- **Write-path impossibility of "cheap" rewrite.** Because publication + stages whole temporary files even when content exists, the exclusive + write cost of touching an existing shared trunk chunk is its full encoded + size (section 9). Admission policy that avoids re-staging unchanged + content is therefore the cheapest SSD lever available; the connector's + `store_skipped_present` and quorum counters already remove most of this + before the write path, which is the implemented (non-heat) baseline. + +First-publisher attribution: the publisher that first creates an object pays +its host-write cost; later referencing roots pay zero retained bytes for it. +The ledger attributes marginal bytes to each root as defined above, which is +the quantity an eviction decision needs. + +## 7. TinyLFU shadow evaluation + +The design evaluates frequency-aware admission *off the serving path* by +replaying traces through a shadow instance — a bounded, pure-Python cache +that mirrors the resident set under a candidate admission policy and reports +the decisions it would have made. + +Structure (W-TinyLFU-style, simplified): + +1. **Window** (`window_capacity`, default 1024 entries, plain LRU): every + distinct-key miss passes through the window before touching the main + cache, which separates one-shot scans from reusable contexts. +2. **Main cache** (`main_capacity`, default 65,536 entries, plain LRU in the + prototype; the full design segments it into a protected/probationary SLRU + pair — the prototype's single-band simplification is stated explicitly + and only weakens discrimination among main-cache residents). +3. **Admission comparison**: when the window overflows, the evicted window entry + is admitted to the main cache iff the main cache has spare capacity or + the adversary comparison `estimate(candidate) > estimate(victim_lru)` + holds, where both estimates come from the section 3 ring with the same + decay discipline. On a loss, the victim stays and the candidate is + dropped (no ghost-band admission in the prototype). +4. **Sketch discipline**: BLAKE2b-64bit-indexed 8-bit ring as in section 3, + incremented on every access (resident or not), decayed by the shared + window. + +Deviations from textbook W-TinyLFU, deliberate and stated: single-band main +cache (no SLRU segmentation), no ghost admission on rejection, no Cuckoo +filter (a direct-mapped ring substitues). They bias the shadow toward +slightly *pessimistic* hit retention relative to full W-TinyLFU. + +Decision output per access, as `ShadowDecision`: + +| reason | meaning | +|---|---| +| `resident_window` | key was resident in the window | +| `resident_main` | key was resident in the main cache (a shadow hit) | +| `spare_capacity` | main cache had room; admitted | +| `admission_win` | evicted the main-cache victim; admitted | +| `admission_loss` | candidate estimate at or below the victim; rejected | + +`evaluate_trace(keys)` consumes an iterable without retaining every decision +and summarizes the trace into a +`TraceReport` (`requests`, `window_hits`, `main_hits`, `misses`, `admitted`, +`rejected`, `hit_rate`). Comparing `hit_rate` for (a) unlimited-cache +replay and (b) shadow admission replay against a recorded production hit +rate is the experiment this design exists to run before any connector +wiring. A deployment-integrable policy additionally needs quorum-aware +cohort decisions across ranks: an admission must hold for all physical +ranks or none (research-only prerequisite listed in section 12). + +Status: **implemented** as a research prototype; integration is +**research-only**; the shadow never affects production serving (**unsupported** +by design, per the isolation contract). + +## 8. Hourly and daily staged-write budget simulation + +A **logical retained byte** is a byte of encoded durable state that became +newly reachable under a cache root. +Events are recorded per publication with the fields of schema +`sparkcache-research-write-event/v1`: + +```json +{ + "schema": "sparkcache-research-write-event/v1", + "at_ns": 1756425600000000000, + "kind": "commit", + "storage_key": "<64 hex characters>", + "context_digest": "<64 hex characters>", + "unique_object_bytes": 183014, + "staged_write_bytes": 183014 +} +``` + +- `kind` is one of `commit` (exact chunks plus manifest), `alias_publication` + (alias files plus added descriptor segments), `metadata_touch` (recency + metadata), or `repair` (invalidation-driven republish). +- `unique_object_bytes` is the encoded size of objects that did not exist + before publication and remain reachable afterward. Serving publication reports + expose `committed_unique_object_bytes` for this quantity. Do not substitute + `CommitReceipt.encoded_bytes`, which includes referenced objects that may + already exist. The research ledger does not ingest serving reports automatically. +- `staged_write_bytes` counts payload bytes passed to temporary-file writes, + including re-staging of identical content (section 9). The prototype + requires the caller to supply this value. Serving reports already expose + `staged_write_bytes`; connecting those reports to the ledger remains unsupported. + +Windows are UTC-aligned from absolute nanoseconds: the hourly window index +is `at_ns // 3_600_000_000_000` and the daily index is +`at_ns // 86_400_000_000_000`. `WriteLedger.hourly_reports(budget)` and +`.daily_reports(budget)` fold events into `BudgetReport` rows per window: + +```json +{ + "window_start_ns": 1756425600000000000, + "window_end_ns": 1756429200000000000, + "unique_object_bytes": 1245583360, + "staged_write_bytes": 2242054400, + "limit_bytes": 2000000000, + "exceeded": true, + "over_bytes": 242054400, + "events": 97 +} +``` + +Budget limits are `WriteBudget(hourly_limit_bytes=None, +daily_limit_bytes=None)`; `None` means monitored-not-limited and sets +`exceeded = None`. A configured limit applies to `staged_write_bytes`, the +prototype's closest in-process measure of write-path pressure. `exceeded` is +a reported fact about a completed window. +Nothing in the prototype enforces anything: enforcement — declining or +deferring a publication whose projected commit would exceed a budget — is +**unsupported**. Enforcement would need: a pre-commit projection API on +`ManifestTransaction` (transactions expose no projected-total +query), a decision point that cannot block a serving thread (violating +"serve never waits" is the known risk), and a defined degradation (skip the +store cleanly, like `store_skipped_busy` does). Those prerequisites are +listed in section 11; until they exist, budgets are reports only. + +## 9. Logical versus physical write amplification + +Three byte quantities are distinct and all three are measured or modeled: + +| Quantity | Meaning | Source | +|---|---|---| +| `unique_object_bytes` | bytes of newly retained durable state | caller-supplied events; serving reports expose `committed_unique_object_bytes` | +| `staged_write_bytes` | payload bytes pushed through temporary-file writes, including staging of identical content | caller-supplied events; serving reports expose `staged_write_bytes` | +| `host_written_bytes` | device-side counter of host writes over an interval | NVMe Data Units Written delta (section 10) | + +Derived ratios the prototype reports: + +``` +staging_ratio = staged_write_bytes / unique_object_bytes +host_ratio = host_written_bytes / unique_object_bytes # the meaningful WAF proxy +``` + +Why they diverge, mechanically, in this codebase: + +1. **Identical-content re-staging.** `_publish_immutable` and + `_publish_immutable_batch` always write a `.writing-` temporary + before attempting the link. When the destination already exists with + identical bytes (the common case when a quorum loser re-publishes, or + when a grown conversation's earlier chunks are re-committed), the bytes + are written, verified, discarded, and the link is a metadata op. The + retained-log accounting shows zero; the device counter does not. The + scheduler-side dedup (`store_skipped_present`, quorum short-circuits) + keeps most of this off the write path already; a heat-independent + improvement would be probing destination existence before staging — + not proposed here, only named as the mechanism. +2. **Allocation rounding and metadata.** Each object rounds up to whole + allocation blocks (4 KiB-programmed bound in + `allocated_bytes_upper_bound`; filesystem-dependent in measurement), and + directory fsyncs plus temp-file create/unlink cycles contribute + bookkeeping blocks invisible to logical accounting. +3. **Alias publication.** Publishing up to 64 aliases plus descriptor + segments over one exact root is small logically (segment files are ~16 + descriptors of small JSON) but multiplies file count; with block + rounding each segment file dominates its content. Device amplification + from alias publication concentrates in block-rounding, measured only + through `host_ratio`. +4. **Off-cache writes share the device.** `host_ratio` computed against + cache-ledger bytes is valid only within a controlled interval where the + workload's non-cache writes are known to be zero or bounded. Otherwise + the ratio is an upper bound and must be reported as such. + +`write_amplification(unique_object_bytes, staged_write_bytes, +host_written_bytes)` returns a `WriteAmplificationEstimate` with the two +ratios and explains missing inputs as `None` rather than substituting +guesses. Physical (media-level) amplification — NAND writes versus host +writes, garbage collection effects — is **unsupported**: the standard Data +Units Written counter is a host-interface counter and deliberately does not +report media amplification; a device-specific endurance telemetry field +would be required and none is consumed here. + +## 10. NVMe Data Units Written monitoring + +The field layout and counter semantics below follow the +[NVM Express Base Specification 2.3](https://nvmexpress.org/wp-content/uploads/NVM-Express-Base-Specification-Revision-2.3-2025.08.01-Ratified.pdf), +SMART / Health Information log. The log is 512 bytes and may be obtained, for example, +as the output of `nvme smart-log /dev/nvme0` on a system with `nvme-cli`, +or the same log page read programmatically) contains at fixed offsets: + +| Offset | Size | Field | +|---|---|---| +| 0x00 | 1 | Critical Warning (bit flags) | +| 0x02 | 2 | Composite Temperature (kelvin, little-endian) | +| 0x04 | 1 | Available Spare | +| 0x05 | 1 | Available Spare Threshold | +| 0x06 | 1 | Percentage Used | +| 0x20 | 16 | Data Units Read (128-bit little-endian) | +| 0x30 | 16 | Data Units Written (128-bit little-endian) | + +Data Units Written counts host writes in units of 1,000 512-byte data +units, rounded up, metadata excluded; the count is converted to 512-byte +units regardless of the namespace LBA size. Therefore: + +``` +host_written_bytes_estimate = data_units_written_units * 512_000 +``` + +The counter is quantized and rounded up, so a delta is an estimate rather than +an exact byte count (the unit constant is `DUW_UNIT_BYTES` in the prototype). +A reported value of +`0` means "not reported" per the specification, so `SmartHealthSample` +carries `data_units_written_reported = units != 0` and practitioners must +treat a zero as missing rather than as zero bytes written. + +The prototype `parse_smart_log_page(page)` consumes the raw 512 bytes, +validates the length and the two 128-bit fields, and returns a +`SmartHealthSample` at an operator-supplied `at_ns`. +`DuwMonitor.delta(first, second)` yields `DuwDelta(units, bytes_est, +seconds, rate_bytes_per_second)` and raises `ResearchFormatError` naming +its cause when either counter is unreported, the counter decreases (device +replacement, counter reset, or samples from different devices), or the +samples carry conflicting non-empty device identifiers. + +Monitoring procedure (operator-executed, no connector involvement): + +1. Sample on a cadence with a stable clock (daily is adequate; hourly if + correlating against `WriteLedger` windows). +2. Persist `sample_to_json(sample)` rows, schema + `sparkcache-research-ssd-sample/v1`. +3. Compare each delta's `bytes_est` against the matching + `daily_reports().unique_object_bytes` (and `staged_write_bytes`) — + `host_ratio` from section 9 is the resulting write-amplification proxy. +4. Relate deltas to endurance: a delta's bytes over the device's rated + TBW gives a consumed-endurance fraction; `Percentage Used` (offset + 0x06) is the vendor's own normalization and should be checked for + drift against the computed fraction. + +Limitations, stated: the counter covers the whole device, not a directory; +multi-tenant devices need controlled intervals; it excludes metadata +host-side yet includes file-system metadata writes, so device attribution +is always approximate; and acquiring the log page itself (ioctl plumbing, +out-of-band `smartctl` parsing) is intentionally outside the prototype. + +## 11. Offline prototype + +`research/heat_ssd_control/` holds the prototype. Contract: + +- **Excluded from serving packages.** `pyproject.toml` includes only + `sparkcache*` packages, so the research package is absent from wheels and + runtime images. No production module imports it, no configuration option + constructs it, and `import sparkcache` does not import it. +- **No serving-code dependency in either direction.** The modules import + nothing from `sparkcache`, and the production-import isolation regression + parses every Python module under `sparkcache/` to reject a reverse import. + Chunk geometry (256 tokens) is a caller-supplied + parameter whose value must match `CacheIdentity.chunk_tokens` for the + modeled root; the modules do not read it from the codebase. +- **Side-effect free.** No filesystem, network, logging, thread, or + subprocess use. All state is explicit constructor/method arguments. +- **No vllm or torch imports**, keeping the modules GPU-free-runnable. + +Files: + +| File | Contents | +|---|---| +| `research/heat_ssd_control/__init__.py` | Package marker restating the offline isolation contract | +| `research/heat_ssd_control/heat_model.py` | `ResearchFormatError`, `HeatKey`, `HitRing` (8-bit ring, saturation, epoch decay, snapshots), `recomputation_tokens_avoided`, `PublishedContext`, `ChunkLedger` (reference counts, shared-trunk reports, marginal bytes) | +| `research/heat_ssd_control/shadow_admission.py` | `ShadowConfig`, `TinyLFUShadow`, `ShadowDecision`, `TraceReport`, `evaluate_trace` | +| `research/heat_ssd_control/write_budget.py` | `DUW_UNIT_BYTES`, `WriteEvent`, `event_to_json`, `WriteBudget`, `WriteLedger`, `BudgetReport`, `write_amplification`, `parse_smart_log_page`, `SmartHealthSample`, `DuwMonitor`, `DuwDelta`, sample JSON schema `sparkcache-research-ssd-sample/v1` | +| `research/heat_ssd_control/test_prototype.py` | GPU-free behavior, malformed-input, atomic-ledger, and production-import isolation regressions | + +Rejection behavior across all modules: + +- Malformed external inputs — digest shape, negative counts, misaligned byte + arrays, wrong schema strings, short log pages, or a counter reset in a + delta — raise `ResearchFormatError` (a `ValueError` subclass) with a + message naming the field. Nothing is retried, repaired, logged, or + swallowed. +- Unknown-but-parseable inputs are still rejected when the schema demands + exact keys; forward-compatibility guessing is deliberately absent so a + modeling mistake surfaces immediately. +- Counter decrease, conflicting device identity, and duplicate-but-different + publications raise rather than silently overwrite: a model that disagrees + with its inputs must stop, not reconcile. + +Example, end to end with no serving dependency: + +```python +from research.heat_ssd_control.heat_model import HeatKey, HitRing +from research.heat_ssd_control.shadow_admission import TinyLFUShadow, ShadowConfig +from research.heat_ssd_control.write_budget import ( + DUW_UNIT_BYTES, parse_smart_log_page, DuwMonitor, WriteEvent, WriteLedger, + WriteBudget, +) + +key = HeatKey("a" * 64, "b" * 64) +ring = HitRing() +ring.record_hit(key) +assert ring.estimate(key) == 1 + +shadow = TinyLFUShadow(ShadowConfig(window_capacity=4, main_capacity=4)) +trace = [key] * 4 + [HeatKey("a" * 64, "c" * 64)] * 2 + [key] +report = shadow.evaluate_trace(trace) +assert report.requests == 7 and report.hit_rate > 0 + +event = WriteEvent(at_ns=0, kind="commit", storage_key=key.storage_key, + context_digest=key.context_digest, + unique_object_bytes=1000, staged_write_bytes=1000) +ledger = WriteLedger(); ledger.add(event) +report = ledger.hourly_reports(WriteBudget(hourly_limit_bytes=500)) +assert report[0].exceeded + +# sample = operator-supplied 512-byte SMART/Health log page +# delta = DuwMonitor.delta(parse_smart_log_page(page_a, at_ns=t0), +# parse_smart_log_page(page_b, at_ns=t1)) +# delta.bytes_est == delta.units * DUW_UNIT_BYTES +``` + +## 12. Integration prerequisites + +Any change that makes serving consume this design requires, in addition to +the GPU-free prototype tests: + +1. A quorum-consistent admission point: publication candidates must be + accepted or rejected identically across every physical rank, or the + verified-or-recompute contract must tolerate rank-divergent publications + (partial publication is invisible until + `commit_manifest`, so a divergent rejection simply produces no manifest — + that direction is safe, divergent *acceptance* is not). +2. A pre-commit projection hook on `ManifestTransaction` for budget checks, + plus a defined skip-publication degradation that keeps store failures + non-fatal in the same way as other optional publication work. +3. A persistence decision for counters (or documented cold start), resolved + against the isolation contract's prohibition of heat in authenticated + surfaces: if heat is persisted, its files live outside + `_CACHE_DATA_DIRECTORIES` and clear-once semantics must be extended + explicitly. +4. A matched workload comparison measuring time to first token, decode throughput, + publication rate, and write volume with the policy enabled and disabled. + Specify acceptable tradeoffs before changing serving defaults. + +## 13. GPU-free validation + +The offline prototype was validated on CPython 3.12.10 without CUDA, vLLM, +or torch: + +| Command | Result | Conclusion | +|---|---:|---| +| `python -m pytest research -q` | 22 passed | Prototype behavior and serving-package isolation are covered | +| `python -m pytest sparkcache -q` | 738 passed, 7 skipped | The prototype does not change the deployed SparkCache source contract | +| `python -m pytest deploy -q` | 108 passed, 1 skipped | Deployment profile and source-hash checks remain unchanged | +| `python -m ruff check .` | passed | Repository lint rules pass | diff --git a/research/__init__.py b/research/__init__.py new file mode 100644 index 0000000..59dc700 --- /dev/null +++ b/research/__init__.py @@ -0,0 +1 @@ +"""Offline research prototypes that are excluded from SparkCache packages.""" diff --git a/research/heat_ssd_control/__init__.py b/research/heat_ssd_control/__init__.py new file mode 100644 index 0000000..7684b4e --- /dev/null +++ b/research/heat_ssd_control/__init__.py @@ -0,0 +1,11 @@ +"""SparkCache heat-aware admission and SSD-control prototype. + +**Research-only. Offline and excluded from SparkCache packages.** + +No production SparkCache module imports this package, and this module imports +nothing from SparkCache production code. It models heat counters, +shadow admission, and write-budget accounting entirely in process memory with +no I/O and no serving dependency. Constructing any object here has no effect +on storage or serving behavior. GPU-free behavior and import-isolation +regressions live in ``research/heat_ssd_control/test_prototype.py``. +""" diff --git a/research/heat_ssd_control/heat_model.py b/research/heat_ssd_control/heat_model.py new file mode 100644 index 0000000..7864713 --- /dev/null +++ b/research/heat_ssd_control/heat_model.py @@ -0,0 +1,398 @@ +"""Bounded 8-bit hit counters and chunk reference accounting. + +Process-local, in-memory modeling of per-context heat (verified-restore +frequency) and of the byte cost a stored root's removal would reclaim. No +disk state, no locks, and no imports from the serving package. +""" + + +import binascii +import hashlib +import json +from collections import Counter +from dataclasses import dataclass + +RING_SCHEMA = "sparkcache-research-heat-ring/v1" +_DIGEST = 64 +_KEY_SEPARATOR = 0x0A # domain separator between storage_key and context_digest bytes + + +class ResearchFormatError(ValueError): + """A modeling input violates its schema or bounded-domain rule. + + Callers must stop on this error: nothing here contradicts serving state, + so a disagreement means the caller's data or assumptions are wrong. + """ + + +def require_digest(value: str, field: str) -> None: + """Reject any value that is not a 64-character lowercase hex digest.""" + if not isinstance(value, str) or len(value) != _DIGEST or not all( + character in "0123456789abcdef" for character in value + ): + raise ResearchFormatError(f"{field} must be a 64-character lowercase SHA-256 hex digest") + + +_require_digest = require_digest + + +def recomputation_tokens_avoided( + committed_tokens: int, + num_computed_tokens: int, +) -> int: + """Prefill tokens replaced by one completed, verified restore.""" + if ( + type(committed_tokens) is not int + or type(num_computed_tokens) is not int + or committed_tokens < 0 + or num_computed_tokens < 0 + or num_computed_tokens > committed_tokens + ): + raise ResearchFormatError( + "token counts must be non-negative integers with " + "num_computed_tokens <= committed_tokens" + ) + return committed_tokens - num_computed_tokens + + +@dataclass(frozen=True, order=True) +class HeatKey: + """Identity of one stored root: cache identity namespace plus context digest.""" + + storage_key: str + context_digest: str + + def __post_init__(self) -> None: + _require_digest(self.storage_key, "storage_key") + _require_digest(self.context_digest, "context_digest") + + def slot(self, capacity: int) -> int: + """Ring index for this key in one ring of ``capacity`` slots.""" + if type(capacity) is not int or capacity <= 0 or capacity & (capacity - 1): + raise ResearchFormatError("capacity must be a positive power of two") + digest = hashlib.blake2b( + self.storage_key.encode("ascii") + + bytes((_KEY_SEPARATOR,)) + + self.context_digest.encode("ascii"), + digest_size=8, + ).digest() + mask = capacity - 1 + return int.from_bytes(digest, "little") & mask + + +@dataclass(frozen=True) +class HitRingConfig: + capacity: int = 131072 + decay_window: int = 8192 + decay_shift: int = 1 + + def __post_init__(self) -> None: + if ( + type(self.capacity) is not int + or self.capacity <= 0 + or self.capacity & (self.capacity - 1) + ): + raise ResearchFormatError("capacity must be a positive power of two") + if type(self.decay_window) is not int or self.decay_window <= 0: + raise ResearchFormatError("decay_window must be at least 1") + if type(self.decay_shift) is not int or not 0 <= self.decay_shift <= 7: + raise ResearchFormatError("decay_shift must be in [0, 7]") + if self.decay_shift > self.decay_window // 2: + raise ResearchFormatError( + "decay_shift must be much smaller than decay_window or the sketch collapses" + ) + + +class HitRing: + """Fixed-size ring of saturating 8-bit access counters with epoch decay. + + One slot per ring index; distinct contexts may share a slot (false + sharing) and their estimates add. Estimates are comparative inputs for + admission experiments only; per the heat-isolation contract no + correctness decision may consume them. + """ + + def __init__(self, config: HitRingConfig | None = None) -> None: + self._config = config or HitRingConfig() + self._counts = bytearray(self._config.capacity) + self._since_decay = 0 + + @property + def config(self) -> HitRingConfig: + return self._config + + def record_hit(self, key: HeatKey) -> int: + """Increment one key's counter and run one decay sweep when due.""" + if not isinstance(key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + index = key.slot(self._config.capacity) + self._since_decay += 1 + self._maybe_decay() + current = self._counts[index] + if current < 0xFF: + self._counts[index] = current + 1 + return self._counts[index] + + def estimate(self, key: HeatKey) -> int: + if not isinstance(key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + return self._counts[key.slot(self._config.capacity)] + + def _maybe_decay(self) -> None: + if self._since_decay < self._config.decay_window: + return + shift = self._config.decay_shift + for index in range(self._config.capacity): + self._counts[index] >>= shift + self._since_decay = 0 + + def snapshot(self) -> str: + """Serialize all counters as one ``heat-ring/v1`` JSON document. + + Key identities are not serialized; estimates for keys that re-derive + the same slot survive a reload and all others restart at zero. + """ + document = { + "schema": RING_SCHEMA, + "capacity": self._config.capacity, + "decay_window": self._config.decay_window, + "decay_shift": self._config.decay_shift, + "increments_since_decay": self._since_decay, + "counts_hex": self._counts.hex(), + } + return json.dumps(document, sort_keys=True, separators=(",", ":")) + + @classmethod + def from_json(cls, payload: str | bytes) -> "HitRing": + try: + document = json.loads(payload) + except (TypeError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise ResearchFormatError(f"heat ring snapshot is not JSON: {error}") from error + if not isinstance(document, dict) or set(document) != { + "schema", + "capacity", + "decay_window", + "decay_shift", + "increments_since_decay", + "counts_hex", + }: + raise ResearchFormatError(f"heat ring snapshot must match schema {RING_SCHEMA}") + if document["schema"] != RING_SCHEMA: + raise ResearchFormatError(f"unknown heat ring schema {document['schema']!r}") + config = HitRingConfig( + capacity=document["capacity"], + decay_window=document["decay_window"], + decay_shift=document["decay_shift"], + ) + counts_hex = document["counts_hex"] + expected = config.capacity * 2 + if not isinstance(counts_hex, str) or len(counts_hex) != expected: + raise ResearchFormatError( + f"counts_hex must be {expected} hex characters for capacity {config.capacity}" + ) + try: + counts = binascii.unhexlify(counts_hex) + except (ValueError, binascii.Error) as error: + raise ResearchFormatError(f"counts_hex is not hex: {error}") from error + since_decay = document["increments_since_decay"] + if not isinstance(since_decay, int) or not 0 <= since_decay < config.decay_window: + raise ResearchFormatError("increments_since_decay is out of range") + ring = object.__new__(cls) + ring._config = config + ring._counts = bytearray(counts) + ring._since_decay = since_decay + return ring + + +@dataclass(frozen=True) +class PublishedContext: + """One stored root as the ledger sees it: ordered chunks plus metadata files.""" + + key: HeatKey + chunk_digests: tuple[str, ...] + chunk_bytes: tuple[int, ...] + chunk_token_counts: tuple[int, ...] + manifest_bytes: int + segment_digests: tuple[str, ...] = () + segment_bytes: tuple[int, ...] = () + + def __post_init__(self) -> None: + if not isinstance(self.key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + if len(self.chunk_digests) != len(self.chunk_bytes): + raise ResearchFormatError("chunk digest and byte counts must align") + if len(self.chunk_digests) != len(self.chunk_token_counts): + raise ResearchFormatError("chunk digest and token counts must align") + if len(self.segment_digests) != len(self.segment_bytes): + raise ResearchFormatError("segment digest and byte counts must align") + byte_counts = (self.manifest_bytes, *self.chunk_bytes, *self.segment_bytes) + if any(type(size) is not int or size < 0 for size in byte_counts): + raise ResearchFormatError("byte counts must be non-negative integers") + if any( + type(token_count) is not int or token_count <= 0 + for token_count in self.chunk_token_counts + ): + raise ResearchFormatError("chunk token counts must be positive integers") + if len(set(self.chunk_digests)) != len(self.chunk_digests): + raise ResearchFormatError("chunk digests must be unique within one root") + if len(set(self.segment_digests)) != len(self.segment_digests): + raise ResearchFormatError("segment digests must be unique within one root") + for digest in self.chunk_digests: + _require_digest(digest, "chunk digest") + for digest in self.segment_digests: + _require_digest(digest, "segment digest") + + +@dataclass(frozen=True) +class ContextHeatReport: + """Per-root cost and share facts derived from publication state.""" + + chunk_count: int + shared_chunk_count: int + shared_tokens: int + retained_shared_bytes: int + marginal_bytes: int + encoded_bytes: int + + +class ChunkLedger: + """Reference counts over content-addressed chunks and descriptor segments. + + Mirrors the dedup structure of the production store: one published chunk + is shared by every root whose manifest or alias chain references it. + ``marginal_bytes`` reproduces what ``MaintenanceReport.bytes_reclaimed`` + would measure for removing one root; it never counts a shared object + toward more than one removal. + """ + + def __init__(self, chunk_tokens: int = 256) -> None: + if type(chunk_tokens) is not int or chunk_tokens <= 0: + raise ResearchFormatError("chunk_tokens must be a positive integer") + self._chunk_tokens = chunk_tokens + self._chunk_refs: Counter[str] = Counter() + self._chunk_bytes: dict[str, int] = {} + self._chunk_token_counts: dict[str, int] = {} + self._segment_refs: Counter[tuple[str, str]] = Counter() + self._segment_bytes: dict[tuple[str, str], int] = {} + self._contexts: dict[HeatKey, PublishedContext] = {} + + @property + def chunk_tokens(self) -> int: + return self._chunk_tokens + + def publish(self, context: PublishedContext) -> None: + if not isinstance(context, PublishedContext): + raise ResearchFormatError("context must be a PublishedContext") + if context.key in self._contexts: + raise ResearchFormatError(f"context {context.key} is already published") + # Validate the complete publication before changing reference counts. + # A rejected research input must not leave the ledger half-updated. + for digest, size, token_count in zip( + context.chunk_digests, + context.chunk_bytes, + context.chunk_token_counts, + ): + if token_count > self._chunk_tokens: + raise ResearchFormatError( + f"chunk {digest} exceeds the configured token geometry" + ) + recorded = self._chunk_bytes.get(digest) + if recorded is not None and recorded != size: + raise ResearchFormatError( + f"chunk {digest} byte count changed across publications" + ) + recorded_tokens = self._chunk_token_counts.get(digest) + if recorded_tokens is not None and recorded_tokens != token_count: + raise ResearchFormatError( + f"chunk {digest} token count changed across publications" + ) + namespace = context.key.storage_key + for digest, size in zip(context.segment_digests, context.segment_bytes): + reference = (namespace, digest) + recorded = self._segment_bytes.get(reference) + if recorded is not None and recorded != size: + raise ResearchFormatError( + f"segment {digest} byte count changed across publications" + ) + + for digest, size, token_count in zip( + context.chunk_digests, + context.chunk_bytes, + context.chunk_token_counts, + ): + self._chunk_bytes.setdefault(digest, size) + self._chunk_token_counts.setdefault(digest, token_count) + self._chunk_refs[digest] += 1 + for digest, size in zip(context.segment_digests, context.segment_bytes): + reference = (namespace, digest) + self._segment_bytes.setdefault(reference, size) + self._segment_refs[reference] += 1 + self._contexts[context.key] = context + + def remove(self, key: HeatKey) -> None: + if not isinstance(key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + context = self._contexts.pop(key, None) + if context is None: + raise ResearchFormatError(f"context {key} is not published") + for digest in context.chunk_digests: + self._chunk_refs[digest] -= 1 + if self._chunk_refs[digest] == 0: + del self._chunk_bytes[digest] + del self._chunk_token_counts[digest] + del self._chunk_refs[digest] + namespace = key.storage_key + for digest in context.segment_digests: + reference = (namespace, digest) + self._segment_refs[reference] -= 1 + if self._segment_refs[reference] == 0: + del self._segment_bytes[reference] + del self._segment_refs[reference] + + def contexts(self) -> tuple["PublishedContext", ...]: + """Every published context in publication order.""" + return tuple(self._contexts.values()) + + def report(self, key: HeatKey) -> ContextHeatReport: + if not isinstance(key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + context = self._contexts.get(key) + if context is None: + raise ResearchFormatError(f"context {key} is not published") + shared_bytes = 0 + marginal_bytes = context.manifest_bytes + shared_count = 0 + shared_tokens = 0 + for digest, size, token_count in zip( + context.chunk_digests, + context.chunk_bytes, + context.chunk_token_counts, + ): + if self._chunk_refs[digest] >= 2: + shared_count += 1 + shared_bytes += size + shared_tokens += token_count + else: + marginal_bytes += size + for digest, size in zip(context.segment_digests, context.segment_bytes): + if self._segment_refs[(context.key.storage_key, digest)] == 1: + marginal_bytes += size + encoded = context.manifest_bytes + sum(context.chunk_bytes) + sum(context.segment_bytes) + return ContextHeatReport( + chunk_count=len(context.chunk_digests), + shared_chunk_count=shared_count, + shared_tokens=shared_tokens, + retained_shared_bytes=shared_bytes, + marginal_bytes=marginal_bytes, + encoded_bytes=encoded, + ) + + +def context_reports(ledger: ChunkLedger) -> dict[HeatKey, ContextHeatReport]: + """Heat reports keyed by complete storage namespace and context identity.""" + if not isinstance(ledger, ChunkLedger): + raise ResearchFormatError("ledger must be a ChunkLedger") + return { + context.key: ledger.report(context.key) + for context in ledger.contexts() + } diff --git a/research/heat_ssd_control/shadow_admission.py b/research/heat_ssd_control/shadow_admission.py new file mode 100644 index 0000000..0f675fa --- /dev/null +++ b/research/heat_ssd_control/shadow_admission.py @@ -0,0 +1,163 @@ +"""TinyLFU-style shadow cache for admission experiments. + +Replays a key sequence through an in-memory two-band cache whose admission +comparison reads the ``HitRing`` frequency sketch. The shadow decides what +frequency-aware admission would have retained; it never serves, never +persists, and shares no state with any serving component. +""" + +from __future__ import annotations + +from collections import OrderedDict +from dataclasses import dataclass +from typing import Iterable + +from research.heat_ssd_control.heat_model import ( + HeatKey, + HitRing, + HitRingConfig, + ResearchFormatError, +) + + +@dataclass(frozen=True) +class ShadowConfig: + window_capacity: int = 1024 + main_capacity: int = 65536 + ring: HitRingConfig | None = None + + def __post_init__(self) -> None: + if ( + type(self.window_capacity) is not int + or type(self.main_capacity) is not int + or self.window_capacity <= 0 + or self.main_capacity <= 0 + ): + raise ValueError("shadow capacities must be positive integers") + if self.ring is not None and not isinstance(self.ring, HitRingConfig): + raise ValueError("ring must be a HitRingConfig or None") + + +@dataclass(frozen=True) +class ShadowDecision: + """What one access would have done under frequency-aware admission.""" + + admitted: bool + hit: bool + reason: str + estimate: int + victim_estimate: int | None + + +@dataclass(frozen=True) +class TraceReport: + """Rollup of one ``evaluate_trace`` run. + + Attributes: + requests: replayed access count. + window_hits: accesses served by the window band. + main_hits: accesses served by the main band; the shadow's hits. + misses: accesses not resident in either band. + admitted: keys that entered or replaced entries in the main cache. + rejected: keys that lost the admission comparison. + final_resident: entries remaining across both bands. + + hit_rate is (window_hits + main_hits) / requests, zero when no + accesses were replayed. + """ + + requests: int + window_hits: int + main_hits: int + misses: int + admitted: int + rejected: int + final_resident: int + + @property + def hit_rate(self) -> float: + if not self.requests: + return 0.0 + return (self.window_hits + self.main_hits) / self.requests + + +class TinyLFUShadow: + """Window + main-cache shadows controlled by the 8-bit heat ring. + + Deliberate simplifications, stated so results are read correctly: + single-band main cache (no protected/probationary segmentation), no + ghost admission on rejection, direct-mapped ring instead of a Cuckoo + filter. Each deviation biases the shadow toward retaining slightly fewer + reusable keys than full W-TinyLFU. + """ + + def __init__(self, config: ShadowConfig | None = None) -> None: + self._config = config or ShadowConfig() + self._ring = HitRing(self._config.ring) + self._window: OrderedDict[HeatKey, None] = OrderedDict() + self._main: OrderedDict[HeatKey, None] = OrderedDict() + + @property + def config(self) -> ShadowConfig: + return self._config + + def access(self, key: HeatKey) -> ShadowDecision: + """Run one access and return the decision it would have produced.""" + if not isinstance(key, HeatKey): + raise ResearchFormatError("key must be a HeatKey") + self._ring.record_hit(key) + if key in self._window: + self._window.move_to_end(key) + return ShadowDecision(True, True, "resident_window", self._ring.estimate(key), None) + if key in self._main: + self._main.move_to_end(key) + return ShadowDecision(True, True, "resident_main", self._ring.estimate(key), None) + + self._window[key] = None + if len(self._window) <= self._config.window_capacity: + return ShadowDecision(False, False, "window", self._ring.estimate(key), None) + + candidate, _ = self._window.popitem(last=False) + if len(self._main) < self._config.main_capacity: + self._main[candidate] = None + return ShadowDecision(True, False, "spare_capacity", self._ring.estimate(candidate), None) + victim = next(iter(self._main)) # least-recently-used main entry + victim_estimate = self._ring.estimate(victim) + candidate_estimate = self._ring.estimate(candidate) + if candidate_estimate > victim_estimate: + self._main.popitem(last=False) + self._main[candidate] = None + return ShadowDecision( + True, False, "admission_win", candidate_estimate, victim_estimate + ) + return ShadowDecision(False, False, "admission_loss", candidate_estimate, victim_estimate) + + def evaluate_trace(self, keys: Iterable[HeatKey]) -> TraceReport: + requests = 0 + window_hits = 0 + main_hits = 0 + misses = 0 + admitted = 0 + rejected = 0 + for key in keys: + decision = self.access(key) + requests += 1 + window_hits += decision.reason == "resident_window" + main_hits += decision.reason == "resident_main" + misses += not decision.hit + admitted += decision.admitted and not decision.hit + rejected += decision.reason == "admission_loss" + return TraceReport( + requests=requests, + window_hits=window_hits, + main_hits=main_hits, + misses=misses, + admitted=admitted, + rejected=rejected, + final_resident=len(self._window) + len(self._main), + ) + + +def evaluate_trace(keys: Iterable[HeatKey], config: ShadowConfig | None = None) -> TraceReport: + """One-shot convenience: fresh shadow, replay, report.""" + return TinyLFUShadow(config).evaluate_trace(keys) diff --git a/research/heat_ssd_control/test_prototype.py b/research/heat_ssd_control/test_prototype.py new file mode 100644 index 0000000..478f7a7 --- /dev/null +++ b/research/heat_ssd_control/test_prototype.py @@ -0,0 +1,367 @@ +"""GPU-free regression coverage for the isolated heat and SSD prototype.""" + +from __future__ import annotations + +import ast +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from research.heat_ssd_control.heat_model import ( + ChunkLedger, + HeatKey, + HitRing, + HitRingConfig, + PublishedContext, + ResearchFormatError, + context_reports, + recomputation_tokens_avoided, +) +from research.heat_ssd_control.shadow_admission import ShadowConfig, TinyLFUShadow +from research.heat_ssd_control.write_budget import ( + DUW_UNIT_BYTES, + DuwMonitor, + WriteBudget, + WriteEvent, + WriteLedger, + event_to_json, + parse_smart_log_page, + sample_to_json, + write_amplification, +) + + +def _digest(character: str) -> str: + return character * 64 + + +def _key(context: str, storage: str = "a") -> HeatKey: + return HeatKey(_digest(storage), _digest(context)) + + +def _context( + context: str, + chunks: tuple[tuple[str, int], ...], + *, + manifest_bytes: int = 10, + segments: tuple[tuple[str, int], ...] = (), + storage: str = "a", + chunk_token_counts: tuple[int, ...] | None = None, +) -> PublishedContext: + return PublishedContext( + key=_key(context, storage), + chunk_digests=tuple(_digest(digest) for digest, _size in chunks), + chunk_bytes=tuple(size for _digest_character, size in chunks), + chunk_token_counts=( + chunk_token_counts + if chunk_token_counts is not None + else (256,) * len(chunks) + ), + manifest_bytes=manifest_bytes, + segment_digests=tuple(_digest(digest) for digest, _size in segments), + segment_bytes=tuple(size for _digest_character, size in segments), + ) + + +def test_importing_sparkcache_does_not_load_offline_research_modules() -> None: + result = subprocess.run( + [ + sys.executable, + "-c", + ( + "import sparkcache,sys; " + "assert not any(name == 'research.heat_ssd_control' or " + "name.startswith('research.heat_ssd_control.') " + "for name in sys.modules)" + ), + ], + check=False, + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr + + +def test_production_modules_do_not_import_research_package() -> None: + package_root = Path(__file__).resolve().parents[2] / "sparkcache" + offenders: list[str] = [] + for path in package_root.rglob("*.py"): + relative = path.relative_to(package_root) + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Import) and any( + alias.name == "research.heat_ssd_control" + or alias.name.startswith("research.heat_ssd_control.") + for alias in node.names + ): + offenders.append(str(relative)) + if isinstance(node, ast.ImportFrom) and node.module and ( + node.module == "research.heat_ssd_control" + or node.module.startswith("research.heat_ssd_control.") + ): + offenders.append(str(relative)) + assert offenders == [] + + +def test_hit_ring_saturates_and_decay_counts_saturated_accesses() -> None: + key = _key("b") + ring = HitRing(HitRingConfig(capacity=8, decay_window=256, decay_shift=1)) + for _ in range(255): + ring.record_hit(key) + assert ring.estimate(key) == 255 + + # The 256th access starts a decayed epoch even though the previous counter + # was saturated: 255 >> 1, followed by the in-flight increment. + assert ring.record_hit(key) == 128 + assert json.loads(ring.snapshot())["increments_since_decay"] == 0 + + +def test_recomputation_tokens_avoided_uses_the_verified_restore_span() -> None: + assert recomputation_tokens_avoided(131_072, 4_096) == 126_976 + with pytest.raises(ResearchFormatError, match="num_computed_tokens"): + recomputation_tokens_avoided(4_096, 4_097) + + +def test_hit_ring_snapshot_round_trip_and_schema_rejection() -> None: + key = _key("b") + ring = HitRing(HitRingConfig(capacity=8, decay_window=16, decay_shift=1)) + ring.record_hit(key) + restored = HitRing.from_json(ring.snapshot()) + assert restored.estimate(key) == 1 + + document = json.loads(ring.snapshot()) + document["counts_hex"] = "00" + with pytest.raises(ResearchFormatError, match="counts_hex"): + HitRing.from_json(json.dumps(document)) + with pytest.raises(ResearchFormatError, match="JSON"): + HitRing.from_json(None) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "factory", + [ + lambda: HeatKey("A" * 64, _digest("b")), + lambda: HitRingConfig(capacity=3), + lambda: HitRingConfig(capacity=True), + lambda: HitRingConfig(decay_window=0), + lambda: HitRingConfig(decay_window=2, decay_shift=2), + lambda: ChunkLedger(chunk_tokens=True), + ], +) +def test_heat_model_rejects_malformed_bounded_inputs(factory: object) -> None: + with pytest.raises(ResearchFormatError): + factory() # type: ignore[operator] + + +def test_chunk_ledger_reports_shared_trunks_and_exclusive_bytes() -> None: + first = _context("b", (("d", 100), ("e", 200)), segments=(("f", 20),)) + second = _context( + "c", + (("d", 100), ("0", 300)), + manifest_bytes=11, + segments=(("f", 20),), + ) + ledger = ChunkLedger(chunk_tokens=256) + ledger.publish(first) + ledger.publish(second) + + first_report = ledger.report(first.key) + assert first_report.shared_chunk_count == 1 + assert first_report.shared_tokens == 256 + assert first_report.retained_shared_bytes == 100 + assert first_report.marginal_bytes == 210 + assert first_report.encoded_bytes == 330 + assert set(context_reports(ledger)) == {first.key, second.key} + + ledger.remove(second.key) + unshared = ledger.report(first.key) + assert unshared.shared_chunk_count == 0 + assert unshared.marginal_bytes == 330 + + +def test_chunk_ledger_uses_exact_shared_terminal_span() -> None: + first = _context("b", (("d", 100),), chunk_token_counts=(128,)) + second = _context("c", (("d", 100),), chunk_token_counts=(128,)) + ledger = ChunkLedger(chunk_tokens=256) + ledger.publish(first) + ledger.publish(second) + assert ledger.report(first.key).shared_tokens == 128 + + +def test_chunk_ledger_rejected_publication_is_atomic() -> None: + existing = _context("b", (("d", 100),)) + conflicting = _context("c", (("e", 50), ("d", 999))) + independent = _context("0", (("e", 50),)) + ledger = ChunkLedger() + ledger.publish(existing) + + with pytest.raises(ResearchFormatError, match="byte count changed"): + ledger.publish(conflicting) + + ledger.publish(independent) + assert ledger.report(independent.key).marginal_bytes == 60 + + +def test_published_context_rejects_duplicate_objects_and_non_integer_bytes() -> None: + with pytest.raises(ResearchFormatError, match="unique"): + _context("b", (("d", 100), ("d", 100))) + with pytest.raises(ResearchFormatError, match="integers"): + _context("b", (("d", True),)) + + +def test_tinylfu_shadow_replays_a_generator_with_bounded_rollup() -> None: + first, second, third = _key("b"), _key("c"), _key("d") + shadow = TinyLFUShadow( + ShadowConfig( + window_capacity=1, + main_capacity=1, + ring=HitRingConfig(capacity=32, decay_window=64, decay_shift=1), + ) + ) + report = shadow.evaluate_trace(key for key in (first, second, first, third, first)) + assert report.requests == 5 + assert report.main_hits == 2 + assert report.window_hits == 0 + assert report.misses == 3 + assert report.admitted == 1 + assert report.rejected == 1 + assert report.final_resident == 2 + assert report.hit_rate == pytest.approx(0.4) + + +def test_write_windows_budget_staged_bytes_not_retained_bytes() -> None: + first = WriteEvent( + at_ns=0, + kind="commit", + storage_key=_digest("a"), + context_digest=_digest("b"), + unique_object_bytes=60, + staged_write_bytes=120, + ) + second = WriteEvent( + at_ns=3_600_000_000_000, + kind="alias_publication", + storage_key=_digest("a"), + context_digest=_digest("c"), + unique_object_bytes=10, + staged_write_bytes=10, + ) + ledger = WriteLedger((first, second)) + reports = ledger.hourly_reports(WriteBudget(hourly_limit_bytes=100)) + assert len(reports) == 2 + assert reports[0].unique_object_bytes == 60 + assert reports[0].staged_write_bytes == 120 + assert reports[0].exceeded is True + assert reports[0].over_bytes == 20 + assert reports[1].exceeded is False + assert ledger.daily_reports()[0].exceeded is None + assert json.loads(event_to_json(first))["schema"] == ( + "sparkcache-research-write-event/v1" + ) + + +def test_write_event_rejects_non_integral_and_impossible_byte_counts() -> None: + common = { + "at_ns": 0, + "kind": "commit", + "storage_key": _digest("a"), + "context_digest": _digest("b"), + } + with pytest.raises(ResearchFormatError, match="integers"): + WriteEvent(**common, unique_object_bytes=True, staged_write_bytes=1) + with pytest.raises(ResearchFormatError, match="cannot exceed"): + WriteEvent(**common, unique_object_bytes=2, staged_write_bytes=1) + + +def test_write_amplification_keeps_missing_denominators_explicit() -> None: + estimate = write_amplification( + unique_object_bytes=100, + staged_write_bytes=150, + host_written_bytes=250, + ) + assert estimate.staging_ratio == 1.5 + assert estimate.host_ratio == 2.5 + + missing = write_amplification(unique_object_bytes=0, staged_write_bytes=10) + assert missing.staging_ratio is None + assert missing.host_ratio is None + + +def _smart_page(*, written_units: int, read_units: int = 0) -> bytes: + page = bytearray(512) + page[0x20:0x30] = read_units.to_bytes(16, "little") + page[0x30:0x40] = written_units.to_bytes(16, "little") + page[0x02:0x04] = (300).to_bytes(2, "little") + page[0x04] = 99 + page[0x05] = 10 + page[0x06] = 7 + return bytes(page) + + +def test_data_units_written_parsing_delta_and_json_schema() -> None: + first = parse_smart_log_page( + _smart_page(written_units=5), at_ns=1_000_000_000, device="/dev/nvme0" + ) + second = parse_smart_log_page( + _smart_page(written_units=9), at_ns=3_000_000_000, device="/dev/nvme0" + ) + delta = DuwMonitor.delta(first, second) + assert delta.units == 4 + assert delta.bytes_est == 4 * DUW_UNIT_BYTES + assert delta.seconds == 2 + assert delta.rate_bytes_per_second == 2 * DUW_UNIT_BYTES + + document = json.loads(sample_to_json(second)) + assert document["schema"] == "sparkcache-research-ssd-sample/v1" + assert document["data_units_written_units"] == 9 + + +def test_data_units_written_rejects_unreported_reset_and_device_mismatch() -> None: + unreported = parse_smart_log_page(_smart_page(written_units=0), at_ns=0) + reported = parse_smart_log_page(_smart_page(written_units=1), at_ns=1) + with pytest.raises(ResearchFormatError, match="not reported"): + DuwMonitor.delta(unreported, reported) + + high = parse_smart_log_page( + _smart_page(written_units=9), at_ns=0, device="/dev/nvme0" + ) + low = parse_smart_log_page( + _smart_page(written_units=8), at_ns=1, device="/dev/nvme0" + ) + with pytest.raises(ResearchFormatError, match="decreased"): + DuwMonitor.delta(high, low) + + other = parse_smart_log_page( + _smart_page(written_units=10), at_ns=1, device="/dev/nvme1" + ) + with pytest.raises(ResearchFormatError, match="conflicting"): + DuwMonitor.delta(high, other) + + +def test_write_ledger_rejects_wrong_type_and_out_of_order_events() -> None: + ledger = WriteLedger() + with pytest.raises(ResearchFormatError, match="WriteEvent"): + ledger.add(object()) # type: ignore[arg-type] + + later = WriteEvent( + at_ns=2, + kind="commit", + storage_key=_digest("a"), + context_digest=_digest("b"), + unique_object_bytes=1, + staged_write_bytes=1, + ) + earlier = WriteEvent( + at_ns=1, + kind="commit", + storage_key=_digest("a"), + context_digest=_digest("c"), + unique_object_bytes=1, + staged_write_bytes=1, + ) + ledger.add(later) + with pytest.raises(ResearchFormatError, match="non-decreasing"): + ledger.add(earlier) diff --git a/research/heat_ssd_control/write_budget.py b/research/heat_ssd_control/write_budget.py new file mode 100644 index 0000000..3e00b6a --- /dev/null +++ b/research/heat_ssd_control/write_budget.py @@ -0,0 +1,384 @@ +"""Logical and host-observed write accounting plus SMART/Health parsing. + +Models the byte quantities the production publication path produces +(immutable-object bytes, staged write-path bytes) and the device-side +NVMe Data Units Written counter, and derives write-amplification ratios +between them. No enforcement: the prototype reports without affecting +publication. +""" + +from __future__ import annotations + +import json + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Sequence + +from research.heat_ssd_control.heat_model import ResearchFormatError, require_digest + +DUW_UNIT_BYTES = 512_000 +"""One reported Data Units Written unit is 1,000 512-byte units, rounded up.""" + +_SMART_LOG_PAGE_BYTES = 512 +_DUW_OFFSET = 0x30 +_DUW_FIELD_BYTES = 16 +_EVENT_KINDS = frozenset( + {"commit", "alias_publication", "metadata_touch", "repair"} +) +_EVENT_SCHEMA = "sparkcache-research-write-event/v1" +_SAMPLE_SCHEMA = "sparkcache-research-ssd-sample/v1" +_HOUR_NS = 3_600_000_000_000 +_DAY_NS = 86_400_000_000_000 + + +@dataclass(frozen=True) +class WriteEvent: + """One publication's byte contribution, as the ledger records it.""" + + at_ns: int + kind: str + storage_key: str + context_digest: str + unique_object_bytes: int + staged_write_bytes: int + + def __post_init__(self) -> None: + if type(self.at_ns) is not int or self.at_ns < 0: + raise ResearchFormatError("at_ns must be a non-negative nanosecond timestamp") + if not isinstance(self.kind, str) or self.kind not in _EVENT_KINDS: + raise ResearchFormatError( + f"kind must be one of {sorted(_EVENT_KINDS)}, got {self.kind!r}" + ) + require_digest(self.storage_key, "storage_key") + require_digest(self.context_digest, "context_digest") + if ( + type(self.unique_object_bytes) is not int + or type(self.staged_write_bytes) is not int + or self.unique_object_bytes < 0 + or self.staged_write_bytes < 0 + ): + raise ResearchFormatError("event byte counts must be non-negative integers") + if self.unique_object_bytes > self.staged_write_bytes: + raise ResearchFormatError( + "unique_object_bytes cannot exceed staged_write_bytes" + ) + + +def event_to_json(event: WriteEvent) -> str: + """Serialize one write event under the ``write-event/v1`` schema.""" + if not isinstance(event, WriteEvent): + raise ResearchFormatError("event must be a WriteEvent") + document = { + "schema": _EVENT_SCHEMA, + "at_ns": event.at_ns, + "kind": event.kind, + "storage_key": event.storage_key, + "context_digest": event.context_digest, + "unique_object_bytes": event.unique_object_bytes, + "staged_write_bytes": event.staged_write_bytes, + } + return json.dumps(document, sort_keys=True, separators=(",", ":")) + + +@dataclass(frozen=True) +class WriteBudget: + """Staged-write interval limits; ``None`` means monitored, not limited.""" + + hourly_limit_bytes: int | None = None + daily_limit_bytes: int | None = None + + def __post_init__(self) -> None: + for field in ("hourly_limit_bytes", "daily_limit_bytes"): + value = getattr(self, field) + if value is not None and (type(value) is not int or value < 0): + raise ResearchFormatError(f"{field} must be None or a non-negative integer") + + +@dataclass(frozen=True) +class BudgetReport: + """One window's totals against its budget.""" + + window_start_ns: int + window_end_ns: int + unique_object_bytes: int + staged_write_bytes: int + limit_bytes: int | None + events: int + + @property + def exceeded(self) -> bool | None: + if self.limit_bytes is None: + return None + return self.staged_write_bytes > self.limit_bytes + + @property + def over_bytes(self) -> int: + if self.limit_bytes is None: + return 0 + return max(0, self.staged_write_bytes - self.limit_bytes) + + +class WriteLedger: + """Accumulates ``WriteEvent`` records and folds them into UTC windows.""" + + def __init__(self, events: Sequence[WriteEvent] = ()) -> None: + self._events: list[WriteEvent] = [] + for event in events: + self.add(event) + + def add(self, event: WriteEvent) -> None: + if not isinstance(event, WriteEvent): + raise ResearchFormatError("event must be a WriteEvent") + if self._events and event.at_ns < self._events[-1].at_ns: + raise ResearchFormatError("events must arrive in non-decreasing at_ns order") + self._events.append(event) + + def hourly_reports(self, budget: WriteBudget | None = None) -> list[BudgetReport]: + if budget is not None and not isinstance(budget, WriteBudget): + raise ResearchFormatError("budget must be a WriteBudget or None") + limit = budget.hourly_limit_bytes if budget else None + return self._fold(_HOUR_NS, limit) + + def daily_reports(self, budget: WriteBudget | None = None) -> list[BudgetReport]: + if budget is not None and not isinstance(budget, WriteBudget): + raise ResearchFormatError("budget must be a WriteBudget or None") + limit = budget.daily_limit_bytes if budget else None + return self._fold(_DAY_NS, limit) + + def _fold(self, window_ns: int, limit: int | None) -> list[BudgetReport]: + reports: dict[int, list[WriteEvent]] = {} + for event in self._events: + reports.setdefault(event.at_ns // window_ns, []).append(event) + return [ + BudgetReport( + window_start_ns=index * window_ns, + window_end_ns=(index + 1) * window_ns, + unique_object_bytes=sum(item.unique_object_bytes for item in events), + staged_write_bytes=sum(item.staged_write_bytes for item in events), + limit_bytes=limit, + events=len(events), + ) + for index, events in sorted(reports.items()) + ] + + +@dataclass(frozen=True) +class WriteAmplificationEstimate: + """Ratios between logical retention, staging, and device counters. + + ``host_ratio`` is valid only over a controlled interval; with any + concurrent non-cache device writes it is an upper bound. + """ + + unique_object_bytes: int | None + staged_write_bytes: int | None + host_written_bytes: int | None + staging_ratio: float | None + host_ratio: float | None + + +def write_amplification( + *, + unique_object_bytes: int | None = None, + staged_write_bytes: int | None = None, + host_written_bytes: int | None = None, +) -> WriteAmplificationEstimate: + for name, value in ( + ("unique_object_bytes", unique_object_bytes), + ("staged_write_bytes", staged_write_bytes), + ("host_written_bytes", host_written_bytes), + ): + if value is not None and (type(value) is not int or value < 0): + raise ResearchFormatError(f"{name} must be None or non-negative") + staging_ratio = ( + staged_write_bytes / unique_object_bytes + if staged_write_bytes is not None and unique_object_bytes + else None + ) + host_ratio = ( + host_written_bytes / unique_object_bytes + if host_written_bytes is not None and unique_object_bytes + else None + ) + return WriteAmplificationEstimate( + unique_object_bytes=unique_object_bytes, + staged_write_bytes=staged_write_bytes, + host_written_bytes=host_written_bytes, + staging_ratio=staging_ratio, + host_ratio=host_ratio, + ) + + +@dataclass(frozen=True) +class SmartHealthSample: + """Parsed SMART/Health log page plus capture metadata.""" + + at_ns: int + device: str + critical_warning: int + composite_temperature_kelvin: int + available_spare: int + available_spare_threshold: int + percentage_used: int + data_units_read_units: int + data_units_written_units: int + data_units_written_reported: bool + + def __post_init__(self) -> None: + if type(self.at_ns) is not int or self.at_ns < 0: + raise ResearchFormatError("at_ns must be a non-negative integer") + if not isinstance(self.device, str): + raise ResearchFormatError("device must be a string") + for field in ( + "critical_warning", + "available_spare", + "available_spare_threshold", + "percentage_used", + ): + value = getattr(self, field) + if type(value) is not int or not 0 <= value <= 0xFF: + raise ResearchFormatError(f"{field} must be an unsigned 8-bit integer") + if ( + type(self.composite_temperature_kelvin) is not int + or not 0 <= self.composite_temperature_kelvin <= 0xFFFF + ): + raise ResearchFormatError( + "composite_temperature_kelvin must be an unsigned 16-bit integer" + ) + for field in ("data_units_read_units", "data_units_written_units"): + value = getattr(self, field) + if type(value) is not int or not 0 <= value < 1 << 128: + raise ResearchFormatError(f"{field} must be an unsigned 128-bit integer") + if type(self.data_units_written_reported) is not bool or ( + self.data_units_written_reported + != (self.data_units_written_units != 0) + ): + raise ResearchFormatError( + "data_units_written_reported must match the nonzero counter" + ) + + @property + def data_units_written_bytes(self) -> int | None: + if not self.data_units_written_reported: + return None + return self.data_units_written_units * DUW_UNIT_BYTES + + +def parse_smart_log_page(page: bytes, *, at_ns: int, device: str = "") -> SmartHealthSample: + """Parse one raw 512-byte SMART/Health log page. + + Acquiring the page (an ``nvme smart-log`` output capture or an ioctl) is + the caller's job; this function interprets the fixed-layout bytes. A + reported Data Units Written value of 0 means "not reported" per the + NVMe specification and is surfaced as ``data_units_written_reported``. + """ + if not isinstance(page, (bytes, bytearray)) or len(page) != _SMART_LOG_PAGE_BYTES: + raise ResearchFormatError( + f"SMART/Health log page must be exactly {_SMART_LOG_PAGE_BYTES} bytes" + ) + if type(at_ns) is not int or at_ns < 0: + raise ResearchFormatError("at_ns must be a non-negative nanosecond timestamp") + if not isinstance(device, str): + raise ResearchFormatError("device must be a string") + + def le128(offset: int) -> int: + chunk = bytes(page[offset : offset + _DUW_FIELD_BYTES]) + return int.from_bytes(chunk, "little") + + return SmartHealthSample( + at_ns=at_ns, + device=device, + critical_warning=page[0x00], + composite_temperature_kelvin=int.from_bytes(page[0x02:0x04], "little"), + available_spare=page[0x04], + available_spare_threshold=page[0x05], + percentage_used=page[0x06], + data_units_read_units=le128(0x20), + data_units_written_units=le128(_DUW_OFFSET), + data_units_written_reported=le128(_DUW_OFFSET) != 0, + ) + + +@dataclass(frozen=True) +class DuwDelta: + """Change of Data Units Written between two same-device samples.""" + + units: int + bytes_est: int + seconds: float + rate_bytes_per_second: float | None + + +class DuwMonitor: + """Compares two parsed SMART/Health samples of one device.""" + + @staticmethod + def delta(first: SmartHealthSample, second: SmartHealthSample) -> DuwDelta: + if not isinstance(first, SmartHealthSample) or not isinstance(second, SmartHealthSample): + raise ResearchFormatError("delta requires two SmartHealthSample inputs") + if first.device and second.device and first.device != second.device: + raise ResearchFormatError( + f"samples carry conflicting device identifiers: {first.device!r} vs {second.device!r}" + ) + if not first.data_units_written_reported or not second.data_units_written_reported: + raise ResearchFormatError( + "Data Units Written is not reported by one or both samples" + ) + if second.at_ns < first.at_ns: + raise ResearchFormatError("second sample precedes the first") + if second.data_units_written_units < first.data_units_written_units: + raise ResearchFormatError( + "Data Units Written decreased between samples: device replacement, " + "counter reset, or samples from different devices" + ) + units = second.data_units_written_units - first.data_units_written_units + seconds = (second.at_ns - first.at_ns) / 1_000_000_000 + if not seconds and units: + raise ResearchFormatError( + "Data Units Written changed without elapsed sample time" + ) + bytes_est = units * DUW_UNIT_BYTES + rate = bytes_est / seconds if seconds > 0 else None + return DuwDelta( + units=units, + bytes_est=bytes_est, + seconds=seconds, + rate_bytes_per_second=rate, + ) + + +def sample_to_json(sample: SmartHealthSample) -> str: + """Serialize one sample under the ``ssd-sample/v1`` schema.""" + if not isinstance(sample, SmartHealthSample): + raise ResearchFormatError("sample must be a SmartHealthSample") + document = { + "schema": _SAMPLE_SCHEMA, + "at_ns": sample.at_ns, + "device": sample.device, + "critical_warning": sample.critical_warning, + "composite_temperature_kelvin": sample.composite_temperature_kelvin, + "available_spare": sample.available_spare, + "available_spare_threshold": sample.available_spare_threshold, + "percentage_used": sample.percentage_used, + "data_units_read_units": sample.data_units_read_units, + "data_units_written_units": sample.data_units_written_units, + "data_units_written_reported": sample.data_units_written_reported, + } + return json.dumps(document, sort_keys=True, separators=(",", ":")) + + +def utc_window_bounds(window_start_ns: int, window_ns: int) -> tuple[str, str]: + """ISO-8601 UTC strings for a window's start and end (reporting aid).""" + if ( + type(window_start_ns) is not int + or type(window_ns) is not int + or window_start_ns < 0 + or window_ns <= 0 + ): + raise ResearchFormatError( + "window_start_ns must be non-negative and window_ns must be positive" + ) + start = datetime.fromtimestamp(window_start_ns / 1_000_000_000, tz=timezone.utc) + end = datetime.fromtimestamp((window_start_ns + window_ns) / 1_000_000_000, tz=timezone.utc) + return start.isoformat(), end.isoformat()