diff --git a/AGENTS.md b/AGENTS.md index b4fa659..2f22bdf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,15 @@ first. the `instance` body field; the topic wins, and `src/control.rs`'s `address()` folds it into the body selector so every handler reads one selector. - Durable **write-ahead** state → crash-safe move+delete with checksum-verify-before-complete. +- **A source-completion failure is a failure, not a success** (DESIGN §20-I, register entry `I`): the + `Verified → CleanupPending → Completed` chain persists the cleanup intent *before* touching the source and + writes `Completed` only once the action is **proven** (archive target present + size/checksum match; delete + source absent). A failed move, a missing/unwritable `archiveDir`, a failed delete, or a mismatched archive + copy leaves the item `CleanupFailed` — no `FileArchived`/`FileDeleted`, no `replicated` increment. Cleanup + retries use the shared full-jitter backoff on their own attempt budget (`retry.maxAttempts`, else 10), + independent of the transfer's time-based `giveUpAfter`; exhaustion emits `FileCleanupFailed` and parks the + item until `trigger` re-drives it. Neither cleanup state is terminal, so a re-discovered source is never + re-enqueued. The `SourceFs` seam in `src/instance/worker.rs` is how the failure paths are tested. - **Long-outage tolerant** (hours–~2d): time-based `giveUpAfter` (default 7d, not attempt caps, shipped) and resume in-flight (shipped). The **disconnection circuit-breaker** (§13.4) is still **not implemented** — `Event::Disconnected`/`Reconnected` exist as enum variants in `src/events.rs` marked `Deferred`, and diff --git a/DESIGN.md b/DESIGN.md index 39a96d1..2a15fbb 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -194,6 +194,7 @@ IDs follow the ecosystem `FR--` convention. RFC-2119 keywords. - **FR-CMP-4** — Completion MUST be **crash-safe** (no loss, no silent double-delivery; idempotent re-verify on restart; §14). - **FR-CMP-5** — `completion` is a **separate instance section**, not part of `egress` (§20-C, accepted). - **FR-CMP-6** — On retry-exhaustion, MUST support **quarantine to a Failed dir** (with an error sidecar) or **retain-in-place** (configurable; §13.3). +- **FR-CMP-7** — A file MUST be recorded as replicated **only when its completion action verifiably happened** (archive target present and matching, or source absent). A failed action is a `CleanupFailed` item, retried on its own bounded budget and surfaced to the operator — never a success (§13.2, §20-I). ### 4.5 Reliability / retry / resume / limits (REL) @@ -512,7 +513,11 @@ stateDiagram-v2 InProgress --> Failed: error (attempt++) Failed --> Ready: backoff elapsed, within giveUpAfter Failed --> Exhausted: giveUpAfter exceeded / maxAttempts hit - Verified --> Completed: completion action (delete or archive) + Verified --> CleanupPending: cleanup intent persisted (write-ahead) + CleanupPending --> Completed: completion action proven (target verified / source gone) + CleanupPending --> CleanupFailed: completion action failed (cleanupAttempt++) + CleanupFailed --> CleanupPending: cleanup backoff elapsed, within the cleanup attempt budget + CleanupFailed --> CleanupPending: trigger (operator re-drive, ignores the gate) Exhausted --> Quarantined: onExhausted=quarantine (move to failedDir) Exhausted --> Retained: onExhausted=retainInPlace Completed --> [*] @@ -520,8 +525,23 @@ stateDiagram-v2 Retained --> [*]: (retriable via trigger) ``` +| State | Meaning | Leaves it | +|---|---|---| +| `Ready` | Readiness passed, durably queued. | A claim (`Ready → InProgress`). | +| `InProgress` | Claimed; delivery to the destinations is underway. | Every destination verified, or an error. | +| `Verified` | Every configured destination delivered **and** integrity-verified. The source is untouched. | The cleanup intent write-ahead. | +| `CleanupPending` | The source completion action (`delete`/`archive`) is authorized but not yet proven done. Persisted **before** the filesystem is touched, so a crash here is re-evaluated against observed state on restart. | The action's proof, or its failure. | +| `CleanupFailed` | Delivered and verified everywhere, but the source was **not** released: the archive move failed, `archiveDir` is missing or unwritable, the delete failed, or the archived copy did not match. Not a success: no `FileArchived`/`FileDeleted`, and `replicated` does not move. | The cleanup backoff gate, or a `trigger`. Once the cleanup attempt budget is spent, only a `trigger`. | +| `Completed` | Terminal success — the source was **verifiably** deleted or archived. | — | +| `Failed` | A transfer attempt errored; awaiting backoff. | The gate elapsing, or the budget expiring. | +| `Exhausted` | `giveUpAfter`/`maxAttempts` spent on the transfer. | `onExhausted`. | +| `Quarantined` | Moved to `failedDir` with an `.error.json` sidecar. | — | +| `Retained` | Left in place, marked failed; retriable via `trigger`. | — | + Every transition is written to the durable store **before** the side effect it authorizes (write-ahead), so -restart re-derives the exact position (§14). +restart re-derives the exact position (§14). `CleanupPending` and `CleanupFailed` are **not** terminal, so a +rescan that re-discovers their still-present source preserves the row instead of re-enqueueing an +already-replicated file as new work. ### 8.2 Per-instance loop @@ -784,16 +804,48 @@ sequenceDiagram D-->>W: delivered with checksum W->>D: verify() D-->>W: ok - W->>S: persist Verified (write-ahead) + W->>S: persist Verified + W->>S: persist CleanupPending (write-ahead) W->>F: completion (delete or move to archive) + F-->>W: target verified / source gone W->>S: persist Completed - Note over W,S: crash between Verified and Completed is recovered idempotently on restart + Note over W,S: crash between CleanupPending and Completed is recovered idempotently on restart + Note over W,S: an unproven action persists CleanupFailed instead — never Completed ``` -**Crash recovery:** a crash between *persist Verified* and *persist Completed* is safe — on restart the item -is re-verified idempotently (the destination object already matches), then the completion action re-runs. The -file is **never re-uploaded and never lost**. Object stores use **stable, deterministic keys** (relpath + -prefix) so re-delivery overwrites identically (idempotent), avoiding duplicates (FR-REL-4). +**Crash recovery:** a crash between *persist CleanupPending* and *persist Completed* is safe — on restart the +item is re-verified idempotently (the destination object already matches), then the completion action is +re-evaluated against observed filesystem state: a source still present means the action is retried, a source +already gone means it landed and the item completes (`move_file` removes the source only after the target +rename succeeds). The file is **never re-uploaded and never lost**. Object stores use **stable, deterministic +keys** (relpath + prefix) so re-delivery overwrites identically (idempotent), avoiding duplicates (FR-REL-4). + +**Completion is proven, not assumed (§20-I).** `Completed` is persisted only once the action is verified: +for `archive`, the target exists at the source's byte count and — under `completion.verify: checksum` — +re-hashes to the checksum the destinations verified against; for `delete`, the source is absent. A failing +move, a missing or unwritable `archiveDir`, a failing delete, or an archived copy that does not match are all +**cleanup failures**: the item becomes `CleanupFailed`, no `FileArchived`/`FileDeleted` is emitted, and the +`replicated` statistic does not move. + +**Cleanup retries** run on their own budget, independent of the transfer's: the shared full-jitter backoff +(`retry.baseDelayMs` → `retry.maxDelayMs`) bounded by `retry.maxAttempts`, or 10 attempts when that is unset. +The transfer's time-based `giveUpAfter` is deliberately not reused — it starts at discovery and is usually +largely spent by the time a slow transfer finishes, which would leave exactly the files that struggled hardest +with no cleanup retries at all. A permanent error (an unconfigured `archiveDir`, a mismatched archive copy) +gives up on the first attempt: no retry can change either. + +**A locked source file is transient on this path.** A `PermissionDenied`/`ResourceBusy` I/O error — Windows +`ERROR_SHARING_VIOLATION` (32), Unix `EBUSY` (16) — usually means a producer, antivirus scanner, indexer, or +backup agent still holds the source open, which clears on its own within seconds. Cleanup retries it under +the backoff rather than parking the file after one attempt. The **transfer** path keeps the opposite rule +(`PermissionDenied` is a credential/ACL problem an operator must fix, so it fails fast to `Exhausted`); the +two classifications are `ReplError::classify_cleanup_io` and `ReplError::classify_io`. + +Every reconciliation tick re-drives the `CleanupPending` rows and the `CleanupFailed` rows whose gate has +elapsed. Once the budget is spent, the item stays `CleanupFailed` with a `FileCleanupFailed` event (§17.1), +a `failed.items[]` entry in `get-status` carrying `state: "cleanup_failed"` and `cleanupAttempts` (§16), and +one way back: fix the cause, then send `trigger`, which re-drives every `CleanupFailed` item regardless of +its gate. ### 13.3 Failure handling & the Failed folder (FR-CMP-6 — your question #9) @@ -1098,11 +1150,18 @@ dispatcher** (resource/verb → handler), structured to be liftable into core al "inProgress":[ { "path": "big.parquet", "size": 734003200, "bytesDone": 220200960, "percent": 30.0, "destination": "s3", "attempt": 1 } ], "replicated":{ "count": 4310, "bytes": 90230411223, "last": { "path": "...", "at": "..." } }, - "failed": { "count": 2, "items": [ { "path": "x.csv", "attempts": 24, "lastError": "...", - "quarantinedAt": "..." } ] } + "failed": { "count": 3, "items": [ { "path": "x.csv", "attempts": 24, "lastError": "...", + "quarantinedAt": "..." }, + { "path": "evidence.bin", "state": "cleanup_failed", + "attempts": 1, "cleanupAttempts": 10, "lastError": "..." } ] } } ``` +The `failed` bucket carries every item needing an operator, discriminated by `items[].state`: +`failed` (retrying), `exhausted`, `quarantined`, and `cleanup_failed` — the last meaning the file reached +every destination but its source was never released (§13.2, §20-I). A `cleanup_failed` row adds +`cleanupAttempts`, because its transfer `attempts` all succeeded. + **P3 reporting notes.** `schedule.mode` reports the **configured** mode verbatim (`immediate` / `cron` / `window`) — never a hardcoded literal — so an instance an operator set to a window is not misrepresented as `immediate` (cron/window *execution* is P4; the window-state sub-fields @@ -1125,9 +1184,14 @@ which derives the `evt/{severity}/{type}` channel from the body — the wire bod ### 17.1 Event types (FR-EVT-1) `FileDiscovered`, `FileReady`, `ReplicationStarted`, `ReplicationProgress`, `ReplicationCompleted`, -`ReplicationFailed`, `FileArchived`, `FileDeleted`, `FileQuarantined`, `RetriesExhausted`, -`ScheduleTriggered`, `WindowOpened`, `WindowClosed`, `ScheduleComplete`, `ScanComplete`, `Disconnected`, -`Reconnected`, **`InstanceActivated`**, **`InstanceDeactivated`**, `ComponentReady`. +`ReplicationFailed`, `FileArchived`, `FileDeleted`, `FileCleanupFailed`, `FileQuarantined`, +`RetriesExhausted`, `ScheduleTriggered`, `WindowOpened`, `WindowClosed`, `ScheduleComplete`, `ScanComplete`, +`Disconnected`, `Reconnected`, **`InstanceActivated`**, **`InstanceDeactivated`**, `ComponentReady`. + +`FileArchived` and `FileDeleted` are emitted only for a completion action that verifiably happened. +`FileCleanupFailed` (`{path, action, attempts}`, last error promoted to `message`, severity `critical`) is +the counterpart: the file replicated and verified on every destination, but its source was not released and +the cleanup attempt budget is spent (§13.2, §20-I). ### 17.2 Example — progress event @@ -1239,6 +1303,38 @@ rehash of the config keys. - **H. UNS shape — REVISED per review:** `{thing}/{component}/{class}/{resource…}` — dropped `edgecommons` root, `v1` (envelope carries version), and `site`/`enterprise` (unreliable tags → envelope); rooted on the IoT-Core-globally-unique `thing`; fits the 256-byte / 7-slash limits (§15). ✔ (confirm §15.2) +- **I. Source-completion failure is a failure, not a success — DECIDED:** a verified delivery whose source + completion action (archive move or delete) does not verifiably happen is recorded as `CleanupFailed`, never + `Completed`. Two durable states carry it — `CleanupPending` (the write-ahead marker persisted *before* the + filesystem is touched) and `CleanupFailed` (the action failed) — and `Completed` is written only once the + action is proven: the archive target exists at the source's byte count and re-hashes to the delivered + checksum under `completion.verify: checksum`, or the deleted source is absent. `FileArchived`/`FileDeleted` + fire only on a proven action, `FileArchived.archivePath` reports the path the file really landed at (which + the `suffix` collision policy can rename) rather than a computed one, and the `replicated` statistic counts + only `Completed`. Cleanup retries use the shared full-jitter backoff on their own attempt budget + (`retry.maxAttempts`, else 10), independent of the transfer's time-based `giveUpAfter`, and a **locked** + source (`PermissionDenied`/`ResourceBusy`, Windows `ERROR_SHARING_VIOLATION`, Unix `EBUSY`) is transient on + this path even though it stays permanent on the transfer path. Exhausted items stay + `CleanupFailed`, emit `FileCleanupFailed`, appear in `get-status` under `failed.items[]` with + `state: "cleanup_failed"` + `cleanupAttempts`, and are re-driven only by `trigger`. §8.1, §13.2, §16, §17.1. + + *Rationale.* The prior behavior logged the filesystem error, persisted `Completed`, counted the file as + replicated, and emitted `FileArchived` with a computed `archivePath` that pointed at a file which was never + written — while the source sat untouched in the watch directory. For an evidence pipeline that is a false + record of custody: the operator's dashboard, the event stream, and the statistics all assert a file was + archived when it was not. A missing `archiveDir` was the same failure by another route, silently degrading + `archive` into "leave it where it is" while still reporting success. The `Completed` marker's original job — + stopping a stale source from being re-discovered in a loop — is preserved by making the two cleanup states + non-terminal: `upsert_ready` only resets a *terminal* row, so a re-discovered `CleanupPending`/`CleanupFailed` + source keeps its row and is never re-enqueued as new work. An attempt-bounded cleanup budget was chosen over + reusing `giveUpAfter` because that clock starts at discovery: a file that spent six days retrying a transfer + would otherwise get no cleanup retry at all. The cleanup path's locked-file rule diverges from the transfer + path deliberately: on egress a `PermissionDenied` is a credential or ACL an operator must fix, but on the + source it is normally a producer, antivirus scanner, indexer, or backup agent still holding the file — the + common case on Windows, and one that clears itself in seconds. Failing fast there would demand operator + action for a self-healing condition. The alternative of a separate top-level `cleanup` section in the + `get-status` document was rejected in favor of the existing `failed` bucket — the item genuinely needs an + operator, `failed.items[].state` already discriminates, and consumers need no new schema. **New decisions in this revision (flag if you disagree):** - Cron-first scheduling (`croner`), English as optional sugar; windows = open+close/duration crons (§12). diff --git a/docs/explanation.md b/docs/explanation.md index 9d5e1bb..ae9913e 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -58,6 +58,51 @@ side effect it authorizes (write-ahead), so a crash between "verified" and "sour idempotently on restart — never re-uploading, never losing the file. Object keys are stable/deterministic so re-delivery overwrites identically. +On restart, a file caught mid-completion is re-evaluated against what is actually on disk rather than +re-applied blindly: a source still present means the completion action runs again, and a source already +gone means it landed, so the file completes. Archiving removes the source only after the target is in +place, which is what makes "the source is gone" a safe signal. + +## Completion is proven, not assumed +A file counts as replicated only when its **source completion action verifiably happened**. After every +destination has delivered and verified, file-replicator archives or deletes the source and then proves it: +under `onSuccess: "archive"` the target must exist at the source's byte count and, with +`completion.verify: "checksum"`, re-hash to the checksum the destinations verified against; under +`onSuccess: "delete"` the source must be gone. + +When the action does not succeed — the archive move fails, `archiveDir` is missing or unwritable, the +delete fails, or the archived copy does not match — the file becomes **`cleanup_failed`**: + +- it is **not** counted in `replicated`, and no `file-archived` or `file-deleted` event is published; +- the source stays where it is, so nothing is lost; +- it appears in `get-status` under `failed.items[]` with `state: "cleanup_failed"` and a + `cleanupAttempts` count. + +The delivered copies are already safe on every destination — what is unresolved is the source. That +distinction matters for evidence and chain-of-custody pipelines, where "archived" appearing in a dashboard +for a file still sitting in the watch directory is worse than an explicit failure. + +Completion attempts retry on their own budget: the same exponential backoff as transfers +(`retry.baseDelayMs` → `retry.maxDelayMs`), bounded by `retry.maxAttempts` or, when that is unset, 10 +attempts. This is separate from the transfer's `retry.giveUpAfter` clock, which starts when the file is +discovered and is often nearly spent by the time a long transfer finishes. Errors that no retry can fix — an +unconfigured `archiveDir`, an archived copy that does not match — stop after the first attempt. + +A **locked source file** is retried, not parked. When a producer, antivirus scanner, indexer, or backup agent +still holds the file open, the delete or move fails with a sharing violation or a permission error that +clears on its own within seconds — the ordinary case on Windows. Completion treats these as transient and +retries them on the backoff above. Transfers keep the opposite rule: there a permission error means a +credential or ACL you have to fix, so it fails fast rather than retrying for days. + +Each reconciliation scan re-drives the files whose completion is still owed. Once a file's completion budget +is spent it stays `cleanup_failed` and publishes a `file-cleanup-failed` event (severity `critical`, with the +`action` that failed and the attempt count). To recover it: fix the cause, then send the `trigger` command, +which re-drives every `cleanup_failed` file regardless of its backoff. + +Because the source of such a file is still in the watch directory, every scan re-discovers it. It is not +re-queued as new work — the bytes already reached every destination, and only the source's release is +outstanding. + ## Scheduling vs windows `cron` releases ready work at each fire; a `window` (open→close cron, or open+duration) gates continuous flow to a time span, for bandwidth conservation. Work outside the window waits; a transfer crossing a diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index 6bb9fe2..00915ee 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -133,6 +133,37 @@ items also show up in `get-status` under `failed.items[]` with `state: "quaranti --- +## Recover files whose source could not be archived or deleted + +A file that reached every destination but whose source could not be released stops in `cleanup_failed`. It +is never reported as replicated, and the source stays in the watch directory. Find these files and clear +them: + +1. Call `get-status` and look for `failed.items[]` entries with `state: "cleanup_failed"`. Each carries + `lastError`, the failing `path`, and `cleanupAttempts`. + + ```jsonc + { "path": "batch-07/evidence.bin", "state": "cleanup_failed", + "attempts": 1, "cleanupAttempts": 10, + "lastError": "permanent: onSuccess=archive requires completion.archiveDir" } + ``` + +2. Fix what `lastError` names. The usual causes are a missing or misspelled `completion.archiveDir`, an + archive volume that is full, read-only, or unmounted, and directory permissions that block the component's + user. + +3. Send the `trigger` command to the instance. It re-drives every `cleanup_failed` file, including the ones + whose completion budget is spent. + +4. Confirm with `get-status`: the files move out of `failed` and into `replicated`, and a `file-archived` or + `file-deleted` event is published for each. + +Subscribe to `file-cleanup-failed` (severity `critical`) to be alerted the moment a file lands in this state +rather than discovering it at the next status poll. See +[explanation › Completion is proven, not assumed](explanation.md#completion-is-proven-not-assumed). + +--- + ## Survive a multi-day disconnection Two mechanisms tolerate an endpoint being down for hours to days — configure the time budget and @@ -194,8 +225,8 @@ RUNNING/STOPPED keepalive). Build a live view by combining two sources: 1. **Subscribe to the event stream** for a device (or the whole fleet): `ecv1/+/FileReplicator/+/evt/#`. Apply each `type` (`file-ready`, `replication-started`, `replication-progress`, `replication-completed`, `replication-failed`, `retries-exhausted`, - `file-archived`/`file-deleted`/`file-quarantined`, …) to your in-memory model. `replication-progress` - carries `percent`/`bytesDone` (throttled). + `file-archived`/`file-deleted`/`file-cleanup-failed`/`file-quarantined`, …) to your in-memory model. + `replication-progress` carries `percent`/`bytesDone` (throttled). 2. **Prime and re-sync with `get-status`** — call it on connect (and periodically) to get the exact `awaiting`/`inProgress`/`replicated`/`failed` document a late subscriber would otherwise have missed. Keep your own timestamped app-layer cache as the retain substitute. diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index c7ea6ff..87cfb1f 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -96,16 +96,20 @@ cron, tz/DST-aware. | Key | Type | Default | Notes | |---|---|---|---| | `onSuccess` | enum | `archive` | `archive` (needs `archiveDir`) \| `delete`. | -| `archiveDir` | string | — | Required when `onSuccess=archive`. | +| `archiveDir` | string | — | Required when `onSuccess=archive`. Without it, no file can be archived: each one is recorded as a cleanup failure rather than a success. | | `onExhausted` | enum | `retainInPlace` | `retainInPlace` \| `quarantine` (needs `failedDir`). | | `failedDir` | string | — | Quarantine dir + `.error.json` sidecar. | | `onCollision` | enum | `suffix` | `suffix` \| `overwrite` \| `fail`. | -| `verify` | enum | `checksum` | `checksum` \| `size` \| `none`. | +| `verify` | enum | `checksum` | `checksum` \| `size` \| `none`. Also decides how the archived copy is proven: `checksum` re-hashes it against the delivered checksum, `size` and `none` check its byte count. | + +The completion action is proven before a file counts as replicated: under `archive` the target must exist +and match, and under `delete` the source must be gone. An action that fails leaves the file in +`cleanup_failed` — see [Explanation - Completion is proven, not assumed](../explanation.md#completion-is-proven-not-assumed). ### `retry` | Key | Type | Default | Notes | |---|---|---|---| -| `baseDelayMs` | int | 1000 | Backoff base. | -| `maxDelayMs` | int | 900000 | Backoff cap (15 min). | -| `giveUpAfter` | string | `7d` | Time budget; governs long-outage tolerance. | -| `maxAttempts` | int | — | Optional hard cap (default: none — time-governed). | +| `baseDelayMs` | int | 1000 | Backoff base. Also the base for source-completion retries. | +| `maxDelayMs` | int | 900000 | Backoff cap (15 min). Also caps source-completion retries. | +| `giveUpAfter` | string | `7d` | Time budget for transfers; governs long-outage tolerance. Source-completion retries are bounded by attempts instead. | +| `maxAttempts` | int | — | Optional hard cap on transfer attempts (default: none — time-governed). It also caps source-completion attempts, which use 10 when it is unset. | diff --git a/docs/reference/data-types.md b/docs/reference/data-types.md index d75db36..d59f122 100644 --- a/docs/reference/data-types.md +++ b/docs/reference/data-types.md @@ -71,11 +71,18 @@ The reply is always wrapped by the command contract: `{ "ok": true, "result": There is no `link` (destination-connectivity) field — file-replicator has no destination circuit-breaker > (see [explanation › Resilience](../explanation.md#resilience-across-long-outages)). Do not depend on it. @@ -125,7 +132,8 @@ object carries the event-specific data; its fields, across all event types, have | `size` | int | `file-ready`, `replication-*` | File size in bytes. | | `destination` | string | `replication-*` | Backend label of the target (`local`/`s3`/…). | | `attempt` | int | `replication-started`/`-progress`/`-failed` | Current attempt, 1-based. | -| `attempts` | int | `retries-exhausted`, `file-quarantined` | Total attempts made. | +| `attempts` | int | `retries-exhausted`, `file-quarantined`, `file-cleanup-failed` | Total attempts made — transfer attempts, except on `file-cleanup-failed`, where it counts source completion attempts. | +| `action` | string | `file-cleanup-failed` | The source completion action that failed: `archive` or `delete`. | | `bytesDone` | int | `replication-progress` | Bytes transferred so far. | | `percent` | float | `replication-progress` | Percent complete, 0.0–100.0. | | `bytes` | int | `replication-completed` | Bytes transferred (total). | @@ -145,7 +153,7 @@ object carries the event-specific data; its fields, across all event types, have | `link` | string | `disconnected` — not emitted | Destination link label (there is no destination circuit-breaker). | **`message` vs `context`.** For the events with a natural error string — `replication-failed`, -`retries-exhausted`, `file-quarantined`, `permission-denied` — the error is promoted to the top-level +`retries-exhausted`, `file-quarantined`, `file-cleanup-failed`, `permission-denied` — the error is promoted to the top-level `message` field and **removed from `context`** (never duplicated in the decoded body). So a consumer reads the human error from `message`, and the machine fields (`path`, `attempt`, `willRetry`, `role`, …) from `context`. diff --git a/docs/reference/messaging-interface.md b/docs/reference/messaging-interface.md index c2afdb4..3163ca4 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -79,6 +79,7 @@ severity + type — the topic and the body can never disagree. Body: | `retries-exhausted` | critical | — | `path`, `destination`, `attempts` (`message` carries the last error) | | `file-archived` | info | — | `path`, `archivePath`? | | `file-deleted` | info | — | `path` | +| `file-cleanup-failed` | critical | — | `path`, `action` (`archive`\|`delete`), `attempts` (`message` carries the last error) | | `file-quarantined` | critical | — | `path`, `attempts`, `quarantinePath`? (`message` carries the last error) | | `scan-complete` | info | — | `discovered`, `awaiting` | | `instance-activated` / `instance-deactivated` | info | — | `source` | @@ -89,6 +90,16 @@ severity + type — the topic and the body can never disagree. Body: | `disconnected` | critical | `raise_alarm` | `link` — not emitted (there is no destination circuit-breaker) | | `permission-denied` | critical | — | `path`, `role` (`ingress`\|`egress`\|`archive`\|`failed`) (`message` carries the error) | +`file-archived` / `file-deleted` — published only for a source completion action that verifiably happened: +the archive target exists and matches the source, or the deleted source is gone. `archivePath` is the path +the file actually landed at, which the `suffix` collision policy can rename. + +`file-cleanup-failed` — the file replicated and verified on every destination, but its source could not be +released and the completion retry budget is spent. The file is **not** counted in `replicated`, its source +stays in the watch directory, and it appears in `get-status` under `failed.items[]` with +`state: "cleanup_failed"`. Fix what `message` reports, then send `trigger` to re-drive it. See +**Completion is proven, not assumed** in `explanation.md`. + `permission-denied` — a directory/target the instance depends on is unreadable/unwritable, at startup or at runtime. ALWAYS emitted, but **deduplicated** so it is not repeated on every rescan or every file: for `ingress`/`archive`/`failed` the dedup key (carried in `context.path`) is the directory, and for `egress` diff --git a/src/control.rs b/src/control.rs index 8f0276e..4aa29e5 100644 --- a/src/control.rs +++ b/src/control.rs @@ -574,6 +574,14 @@ pub(crate) fn instance_state_snapshot( .list_by_state(id, ItemState::Quarantined) .unwrap_or_default(), ); + // `CleanupFailed` belongs here too (DESIGN §20-I): the bytes reached every destination, but the + // source was never released, so the file still needs an operator. It is discriminated by + // `items[].state` and carries `cleanupAttempts`, and it is deliberately absent from `replicated`. + failed.extend( + store + .list_by_state(id, ItemState::CleanupFailed) + .unwrap_or_default(), + ); instance_status_json( &StatusInputs { id, @@ -629,6 +637,11 @@ fn instance_status_json(inp: &StatusInputs, now: i64) -> Value { if w.state == ItemState::Quarantined { m["quarantinedAt"] = json!(rfc3339_ms(w.updated_at)); } + // A cleanup failure's `attempts` are transfer attempts (all of which succeeded), so the + // count that explains the row is the separate source-completion one (DESIGN §20-I). + if w.state == ItemState::CleanupFailed { + m["cleanupAttempts"] = json!(w.cleanup_attempts); + } m }) .collect(); @@ -860,6 +873,7 @@ mod tests { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -948,6 +962,59 @@ mod tests { assert_eq!(v["failed"]["items"][1]["quarantinedAt"], json!("2026-07-01T09:15:22Z")); } + #[test] + fn a_cleanup_failed_item_is_reported_as_needing_attention() { + // DESIGN §20-I: the bytes reached every destination but the source was never released, so the + // file still needs an operator. `state` discriminates it from a transfer failure, and the + // count that explains it is `cleanupAttempts` (its transfer `attempts` all succeeded). + let mut c = work_item("evidence.bin", ItemState::CleanupFailed, 12); + c.attempts = 1; + c.cleanup_attempts = 10; + c.last_error = Some("permanent: onSuccess=archive requires completion.archiveDir".into()); + let failed = vec![c]; + let inp = StatusInputs { + id: "i1", + active: true, + configured_enabled: true, + schedule_mode: "immediate", + dest_label: "local", + stats: Stats { + replicated: 0, + failed: 0, + bytes: 0, + }, + awaiting: &[], + in_progress: &[], + failed: &failed, + }; + let v = instance_status_json(&inp, 100_000); + assert_eq!(v["failed"]["count"], json!(1)); + assert_eq!(v["failed"]["items"][0]["state"], json!("cleanup_failed")); + assert_eq!(v["failed"]["items"][0]["cleanupAttempts"], json!(10)); + assert_eq!( + v["replicated"]["count"], + json!(0), + "a cleanup failure is never counted as replicated" + ); + assert!(v["failed"]["items"][0].get("quarantinedAt").is_none()); + } + + #[test] + fn cleanup_failed_rows_reach_the_status_document_from_the_store() { + // Proves the wiring, not just the serializer: `instance_state_snapshot` must actually query + // the CleanupFailed state alongside failed/exhausted/quarantined. + let store = mem_store(); + store.upsert_ready("i1", "evidence.bin", 12, 0, 1).unwrap(); + store + .set_state("i1", "evidence.bin", ItemState::CleanupFailed, 2) + .unwrap(); + let v = + instance_state_snapshot(store.as_ref(), "i1", true, true, "immediate", "local", 100); + assert_eq!(v["failed"]["count"], json!(1)); + assert_eq!(v["failed"]["items"][0]["path"], json!("evidence.bin")); + assert_eq!(v["failed"]["items"][0]["state"], json!("cleanup_failed")); + } + // ---- addressing (core 0.5.0 scoped commands, D-SC-4) ---------------------------------------- #[test] diff --git a/src/dest/local.rs b/src/dest/local.rs index 5ce81f8..1d7ddaf 100644 --- a/src/dest/local.rs +++ b/src/dest/local.rs @@ -297,6 +297,7 @@ mod tests { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/src/domain.rs b/src/domain.rs index a4796c6..c9113f1 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -21,7 +21,15 @@ pub enum ItemState { InProgress, /// Delivered + integrity-verified; the WRITE-AHEAD point *before* the source side effect (§13.2). Verified, - /// Terminal success — the source has been deleted or archived. + /// Every destination is verified and the source completion action (`delete`/`archive`) is + /// authorized but not yet proven done — the WRITE-AHEAD point *before* the filesystem side effect + /// (DESIGN §13.2/§20-I). Crash recovery re-evaluates it against observed filesystem state. + CleanupPending, + /// Every destination is verified but the source completion action failed. Retried on its own + /// bounded backoff clock; once that budget is spent the item stays here — operator-visible in + /// `get-status`, re-drivable with `trigger`. Never counted as replicated (DESIGN §20-I). + CleanupFailed, + /// Terminal success — the source has been **verifiably** deleted or archived. Completed, /// Last attempt errored; awaiting backoff (attempts recorded, `next_attempt_at` set). Failed, @@ -40,6 +48,8 @@ impl ItemState { ItemState::Ready => "ready", ItemState::InProgress => "in_progress", ItemState::Verified => "verified", + ItemState::CleanupPending => "cleanup_pending", + ItemState::CleanupFailed => "cleanup_failed", ItemState::Completed => "completed", ItemState::Failed => "failed", ItemState::Exhausted => "exhausted", @@ -59,6 +69,8 @@ impl ItemState { "ready" => ItemState::Ready, "in_progress" => ItemState::InProgress, "verified" => ItemState::Verified, + "cleanup_pending" => ItemState::CleanupPending, + "cleanup_failed" => ItemState::CleanupFailed, "completed" => ItemState::Completed, "failed" => ItemState::Failed, "exhausted" => ItemState::Exhausted, @@ -69,6 +81,11 @@ impl ItemState { } /// True for the terminal states no worker will re-claim: `Completed`, `Quarantined`, `Retained`. + /// + /// [`CleanupPending`](Self::CleanupPending) and [`CleanupFailed`](Self::CleanupFailed) are + /// deliberately **not** terminal: the source file is still on disk, so re-discovery must preserve + /// the row (see [`StateStore::upsert_ready`](crate::state::StateStore::upsert_ready), which only + /// resets a *terminal* row) rather than re-enqueue an already-replicated file as new work. pub fn is_terminal(&self) -> bool { matches!( self, @@ -92,8 +109,17 @@ pub struct WorkItem { pub size: u64, /// Unix ms — oldest-ready-first ordering. pub discovered_at: i64, + /// Transfer attempts (the aggregate rollup across destinations). Cleanup attempts are counted + /// separately in [`cleanup_attempts`](Self::cleanup_attempts). pub attempts: u32, - /// Unix ms — backoff gate for re-claim (P1 addition to the §14.2 schema). + /// Source-completion (`delete`/`archive`) attempts made since the item was first verified — the + /// cleanup retry clock, kept independent of the transfer's `attempts`/`giveUpAfter` budget so a + /// file that spent its whole transfer budget still gets a full set of cleanup retries + /// (DESIGN §20-I). + pub cleanup_attempts: u32, + /// Unix ms — backoff gate for re-claim (P1 addition to the §14.2 schema). While the item is + /// `CleanupPending`/`CleanupFailed` the transfer is already done, so this same column carries the + /// **cleanup** retry gate. pub next_attempt_at: i64, pub last_error: Option, pub bytes_done: u64, @@ -224,6 +250,8 @@ mod tests { ItemState::Ready, ItemState::InProgress, ItemState::Verified, + ItemState::CleanupPending, + ItemState::CleanupFailed, ItemState::Completed, ItemState::Failed, ItemState::Exhausted, @@ -262,6 +290,8 @@ mod tests { ItemState::Ready, ItemState::InProgress, ItemState::Verified, + ItemState::CleanupPending, + ItemState::CleanupFailed, ItemState::Failed, ItemState::Exhausted, ] { diff --git a/src/error.rs b/src/error.rs index f47dcd0..569bc69 100644 --- a/src/error.rs +++ b/src/error.rs @@ -101,6 +101,49 @@ impl ReplError { } } + /// Map a raw [`io::Error`] raised by the **source completion action** (`delete`/`archive`, + /// DESIGN §20-I) to a classified [`ReplError`]. Identical to [`classify_io`](Self::classify_io) + /// except that a **locked file** is [`Transient`](Self::Transient) rather than permanent. + /// + /// On the transfer path a `PermissionDenied` means a credential or ACL an operator has to fix, so + /// failing fast is right. On the cleanup path it usually means something else still has the source + /// open — a producer finishing its write, an antivirus scanner, an indexer, a backup agent — which + /// is the normal case on Windows and clears on its own within seconds. Parking such a file in + /// `CleanupFailed` after a single attempt would demand operator action for a condition that heals + /// itself, so these are retried under the cleanup backoff instead. + /// + /// Detection covers both the portable [`io::ErrorKind`] and the platform's raw code, because which + /// of the two a lock surfaces as depends on the OS and the toolchain version (see + /// [`is_locked_file`](Self::is_locked_file)). + pub fn classify_cleanup_io(e: io::Error) -> Self { + if Self::is_locked_file(&e) { + return ReplError::Transient(format!("io: {e} ({:?})", e.kind())); + } + Self::classify_io(e) + } + + /// Whether `e` reports a file another process is holding open (see + /// [`classify_cleanup_io`](Self::classify_cleanup_io)): `PermissionDenied` or `ResourceBusy` by + /// [`io::ErrorKind`], or the platform's raw code — Windows `ERROR_SHARING_VIOLATION` (32), Unix + /// `EBUSY` (16). The raw codes are matched per platform because the same number means something + /// unrelated on the other (32 is `EPIPE` on Unix). + fn is_locked_file(e: &io::Error) -> bool { + if matches!( + e.kind(), + io::ErrorKind::PermissionDenied | io::ErrorKind::ResourceBusy + ) { + return true; + } + #[cfg(windows)] + const LOCKED_RAW: &[i32] = &[32]; // ERROR_SHARING_VIOLATION + #[cfg(unix)] + const LOCKED_RAW: &[i32] = &[16]; // EBUSY + #[cfg(not(any(windows, unix)))] + const LOCKED_RAW: &[i32] = &[]; + e.raw_os_error() + .is_some_and(|code| LOCKED_RAW.contains(&code)) + } + fn io_kind_is_permanent(kind: io::ErrorKind) -> bool { matches!( kind, @@ -175,6 +218,45 @@ mod tests { assert!(!ReplError::Transient("later".into()).is_permission_denied()); } + #[test] + fn classify_cleanup_io_treats_a_locked_file_as_transient() { + // On the CLEANUP path a held-open source (a producer, antivirus scanner, indexer, backup + // agent) must be retried under the cleanup backoff, not parked after one attempt — unlike the + // transfer path, where a permission denial is a credential/ACL problem that fails fast. + for kind in [io::ErrorKind::PermissionDenied, io::ErrorKind::ResourceBusy] { + let e = ReplError::classify_cleanup_io(io::Error::new(kind, "locked")); + assert!(e.is_transient(), "{kind:?} → {e:?}"); + assert!(!e.is_permanent(), "{kind:?} → {e:?}"); + } + // The transfer path is deliberately unchanged. + let transfer = + ReplError::classify_io(io::Error::new(io::ErrorKind::PermissionDenied, "denied")); + assert!(matches!(transfer, ReplError::PermissionDenied(_))); + assert!(transfer.is_permanent()); + } + + #[test] + fn classify_cleanup_io_matches_the_platform_sharing_violation_code() { + // Windows ERROR_SHARING_VIOLATION (32) / Unix EBUSY (16) — matched by raw code as well as by + // `ErrorKind`, because which of the two a lock surfaces as depends on the OS and toolchain. + #[cfg(windows)] + let raw = 32; + #[cfg(not(windows))] + let raw = 16; + let e = ReplError::classify_cleanup_io(io::Error::from_raw_os_error(raw)); + assert!(e.is_transient(), "raw {raw} → {e:?}"); + assert!(!e.is_permanent()); + } + + #[test] + fn classify_cleanup_io_leaves_every_other_kind_alone() { + // Not-found still fails fast (nothing to retry), and an unrelated error stays transient. + let gone = ReplError::classify_cleanup_io(io::Error::new(io::ErrorKind::NotFound, "gone")); + assert!(gone.is_permanent(), "got {gone:?}"); + let slow = ReplError::classify_cleanup_io(io::Error::new(io::ErrorKind::TimedOut, "slow")); + assert!(slow.is_transient(), "got {slow:?}"); + } + #[test] fn classify_io_timeout_is_transient() { let e = ReplError::classify_io(io::Error::new(io::ErrorKind::TimedOut, "slow")); diff --git a/src/events.rs b/src/events.rs index ef414f0..83da5d7 100644 --- a/src/events.rs +++ b/src/events.rs @@ -117,6 +117,16 @@ pub enum Event { }, /// The source file was deleted on success. FileDeleted { path: String }, + /// Every destination is verified but the source completion action (`archive`/`delete`) failed and + /// its retry budget is spent (DESIGN §20-I). The item stays `CleanupFailed` — never `Completed`, + /// never counted as replicated — until an operator fixes the cause and re-drives it with + /// `trigger`. `action` is `"archive"` or `"delete"`. + FileCleanupFailed { + path: String, + action: String, + attempts: u32, + last_error: String, + }, /// The source file was quarantined after exhaustion. `quarantinePath` omitted when unknown. FileQuarantined { path: String, @@ -178,6 +188,7 @@ impl Event { Event::RetriesExhausted { .. } => "RetriesExhausted", Event::FileArchived { .. } => "FileArchived", Event::FileDeleted { .. } => "FileDeleted", + Event::FileCleanupFailed { .. } => "FileCleanupFailed", Event::FileQuarantined { .. } => "FileQuarantined", Event::ScanComplete { .. } => "ScanComplete", Event::InstanceActivated { .. } => "InstanceActivated", @@ -267,6 +278,14 @@ impl Event { m } Event::FileDeleted { path } => obj(json!({ "path": path })), + Event::FileCleanupFailed { + path, + action, + attempts, + last_error, + } => obj(json!({ + "path": path, "action": action, "attempts": attempts, "lastError": last_error + })), Event::FileQuarantined { path, attempts, @@ -321,6 +340,7 @@ impl Event { Event::ReplicationFailed { error, .. } => Some(error.clone()), Event::RetriesExhausted { last_error, .. } => Some(last_error.clone()), Event::FileQuarantined { last_error, .. } => Some(last_error.clone()), + Event::FileCleanupFailed { last_error, .. } => Some(last_error.clone()), Event::PermissionDenied { error, .. } => Some(error.clone()), _ => None, }; @@ -328,6 +348,7 @@ impl Event { Event::ReplicationFailed { .. } => Severity::Warning, Event::RetriesExhausted { .. } | Event::FileQuarantined { .. } + | Event::FileCleanupFailed { .. } | Event::Disconnected { .. } | Event::Reconnected { .. } | Event::PermissionDenied { .. } => Severity::Critical, @@ -344,6 +365,7 @@ impl Event { Event::RetriesExhausted { .. } => "retries-exhausted", Event::FileArchived { .. } => "file-archived", Event::FileDeleted { .. } => "file-deleted", + Event::FileCleanupFailed { .. } => "file-cleanup-failed", Event::FileQuarantined { .. } => "file-quarantined", Event::ScanComplete { .. } => "scan-complete", Event::InstanceActivated { .. } => "instance-activated", @@ -667,6 +689,37 @@ mod tests { assert!(plan.context.get("lastError").is_none(), "promoted to message, not duplicated"); } + #[test] + fn file_cleanup_failed_plan_is_critical_and_promotes_the_last_error() { + // DESIGN §20-I: the file reached every destination, but the source completion action did not + // succeed, so this is a critical operator-facing event — never a `file-archived`/`file-deleted`. + let ev = Event::FileCleanupFailed { + path: "a/b.csv".into(), + action: "archive".into(), + attempts: 4, + last_error: "permanent: onSuccess=archive requires completion.archiveDir".into(), + }; + let plan = ev.plan(); + assert_eq!(plan.severity, Severity::Critical); + assert_eq!(plan.event_type, "file-cleanup-failed"); + assert_eq!( + plan.message, + Some("permanent: onSuccess=archive requires completion.archiveDir".to_string()) + ); + assert_eq!(plan.context["path"], json!("a/b.csv")); + assert_eq!(plan.context["action"], json!("archive")); + assert_eq!(plan.context["attempts"], json!(4)); + assert!( + plan.context.get("lastError").is_none(), + "promoted to message, not duplicated" + ); + assert!( + plan.alarm.is_none(), + "a terminal per-file failure has no clear counterpart" + ); + assert_eq!(ev.name(), "FileCleanupFailed"); + } + #[test] fn optional_paths_omitted_when_absent_present_when_set() { let arch_none = Event::FileArchived { diff --git a/src/instance/mod.rs b/src/instance/mod.rs index 520d314..c489f57 100644 --- a/src/instance/mod.rs +++ b/src/instance/mod.rs @@ -154,6 +154,19 @@ struct ScheduleTransition { window_closed: bool, } +/// The operator overrides a control-plane `trigger` layers on top of an ordinary reconciliation tick +/// (FR-CTL-3 / DESIGN §20-I). The periodic tick uses [`TickForce::default`] — no overrides. +#[derive(Debug, Clone, Copy, Default)] +struct TickForce { + /// Bypass the cron/window schedule gate and drain all ready work now (`trigger`'s + /// `ignoreWindow`). + ignore_window: bool, + /// Re-drive every `CleanupFailed` item, including the ones whose cleanup retry budget is spent — + /// the operator's way back for a file that is replicated everywhere but whose source could not be + /// archived or deleted. + redrive_cleanup: bool, +} + struct QueueMetricSnapshot { ready: usize, values: MetricValues, @@ -457,16 +470,17 @@ impl Instance { /// the schedule (DESIGN §12). A deactivated instance does nothing. `now` is the single Unix-ms /// clock read for this tick. pub async fn tick(&self, now: i64) { - self.run_tick(now, false).await; + self.run_tick(now, TickForce::default()).await; } - /// The tick body. `force = true` is a control-plane `trigger { ignoreWindow: true }` (FR-CTL-3): - /// it still discovers/enqueues and promotes retries as usual, but then **bypasses the schedule - /// gate** and drains ALL ready work now regardless of cron/window state — an explicit operator - /// override. A forced tick does not disturb the cron watermark or window open-state and emits no - /// gate-transition events (the control plane emits its own `ScheduleTriggered`), so it never - /// perturbs the automatic schedule. - async fn run_tick(&self, now: i64, force: bool) { + /// The tick body. [`TickForce`] carries the operator overrides a control-plane `trigger` adds on + /// top of the ordinary reconciliation pass (FR-CTL-3): `ignore_window` still discovers/enqueues and + /// promotes retries as usual but then **bypasses the schedule gate**, draining ALL ready work now + /// regardless of cron/window state, and `redrive_cleanup` re-drives even the `CleanupFailed` items + /// whose cleanup budget is spent (DESIGN §20-I). A forced tick does not disturb the cron watermark + /// or window open-state and emits no gate-transition events (the control plane emits its own + /// `ScheduleTriggered`), so it never perturbs the automatic schedule. + async fn run_tick(&self, now: i64, force: TickForce) { if !self.is_active() { self.emit_schedule_metrics(0, false, false, false, false, 0) .await; @@ -573,13 +587,24 @@ impl Instance { tracing::error!(instance = %self.id, error = %e, "retry promotion task failed") } } + // Cleanup manager (DESIGN §20-I): re-drive every item that is replicated and verified on all + // destinations but whose source completion action has not succeeded — a `CleanupPending` row an + // abort left mid-action, plus the `CleanupFailed` rows whose cleanup backoff gate has elapsed + // (all of them under an operator `trigger`). This runs BEFORE the schedule gate and regardless + // of it: the transfer is already paid for, and a source left un-archived is not something a + // closed replication window should hold hostage. Serialized with the rest of the tick by + // `tick_lock`, so it never races the batch below over the same item. + if let Err(e) = self.worker.drive_cleanup(now, force.redrive_cleanup).await { + tracing::error!(instance = %self.id, error = %e, "cleanup re-drive failed"); + } + let queue_snapshot = self.emit_queue_metrics(now).await; // Scheduling gate (DESIGN §12): decide whether/how ready work may be claimed this tick. // Discovery/enqueue above always ran regardless — only the CLAIM is gated, so newly-ready work // simply accumulates in the durable `Ready` backlog while the gate is closed. A forced trigger // (FR-CTL-3 `ignoreWindow`) bypasses the gate entirely and drains everything now. - let (admission, transition) = if force { + let (admission, transition) = if force.ignore_window { ( Admission::All { drain: true }, ScheduleTransition::default(), @@ -1039,6 +1064,17 @@ impl Instance { ); } } + // Aborted inside the write-ahead cleanup window: the completion action is still + // owed, so drive it here rather than leaving the row non-terminal until the next + // tick's cleanup pass (DESIGN §20-I). + ItemState::CleanupPending => { + if let Err(e) = self.worker.run_cleanup(&row, now).await { + tracing::error!( + instance = %self.id, relpath = %rp, error = %e, + "completing cleanup-pending item after window-close pause failed" + ); + } + } _ => {} }, Ok(None) => {} @@ -1194,7 +1230,19 @@ impl InstanceControl for Instance { // `ignore_window` the tick bypasses the schedule gate so a cron/window instance replicates // now regardless of its schedule (FR-CTL-3); otherwise it is exactly the periodic-rescan tick // (gate-respecting). A deactivated instance's tick is a no-op. - self.run_tick(now, ignore_window).await; + // + // Every `trigger` — with or without `ignoreWindow` — also re-drives the `CleanupFailed` items + // whose cleanup budget is spent (DESIGN §20-I). That is the operator's way back for a file that + // is replicated and verified everywhere but could not be archived or deleted: fix the cause, + // then `trigger`. + self.run_tick( + now, + TickForce { + ignore_window, + redrive_cleanup: true, + }, + ) + .await; } fn apply_activation( @@ -1319,6 +1367,76 @@ mod tests { assert_eq!(store.stats("i1").unwrap().replicated, 2); } + #[tokio::test] + async fn tick_drives_a_cleanup_pending_item_to_completion() { + // DESIGN §20-I: the reconciliation tick owns the cleanup pass, so an item left mid-action by a + // crash or a window-close abort is finished on the next tick rather than sitting non-terminal. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + std::fs::write(src.path().join("a.txt"), b"hello").unwrap(); + let store: Arc = Arc::new(SqliteStore::open_in_memory().unwrap()); + let inst = build( + instance_cfg("i1", src.path(), dst.path(), true), + store.clone(), + ); + // A prior run delivered + verified and wrote the write-ahead cleanup marker, then died. + store.upsert_ready("i1", "a.txt", 5, 0, 1).unwrap(); + store + .set_state("i1", "a.txt", ItemState::CleanupPending, 1) + .unwrap(); + + inst.tick(100).await; + + assert_eq!( + store.get("i1", "a.txt").unwrap().unwrap().state, + ItemState::Completed + ); + assert!(!src.path().join("a.txt").exists(), "source released"); + assert_eq!(store.stats("i1").unwrap().replicated, 1); + } + + #[tokio::test] + async fn a_parked_cleanup_failure_waits_for_a_trigger_not_for_the_next_tick() { + // A cleanup failure whose budget is spent carries a gate no clock reaches, so the ordinary + // tick leaves it alone; `trigger` re-drives it (DESIGN §20-I) — the operator's way back once + // the underlying cause is fixed. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + std::fs::write(src.path().join("a.txt"), b"hello").unwrap(); + let store: Arc = Arc::new(SqliteStore::open_in_memory().unwrap()); + let inst = build( + instance_cfg("i1", src.path(), dst.path(), true), + store.clone(), + ); + store.upsert_ready("i1", "a.txt", 5, 0, 1).unwrap(); + store + .record_cleanup_attempt( + "i1", + "a.txt", + "disk full", + ItemState::CleanupFailed, + i64::MAX, + 1, + ) + .unwrap(); + + inst.tick(100).await; + assert_eq!( + store.get("i1", "a.txt").unwrap().unwrap().state, + ItemState::CleanupFailed, + "the ordinary tick respects the parked gate" + ); + assert!(src.path().join("a.txt").exists()); + + inst.trigger_scan(200, false).await; + assert_eq!( + store.get("i1", "a.txt").unwrap().unwrap().state, + ItemState::Completed, + "trigger re-drives a parked cleanup failure" + ); + assert!(!src.path().join("a.txt").exists()); + } + #[tokio::test] async fn tick_promotes_failed_items_past_their_backoff_gate() { let src = tempfile::tempdir().unwrap(); diff --git a/src/instance/worker.rs b/src/instance/worker.rs index c2c06c5..b906a14 100644 --- a/src/instance/worker.rs +++ b/src/instance/worker.rs @@ -9,7 +9,9 @@ //! └─ error → record per-dest attempt + backoff → retry | dest Exhausted //! //! once EVERY destination is DestPhase::Verified: -//! persist ItemState::Verified (write-ahead) → completion (delete|archive) → persist Completed +//! persist ItemState::Verified → persist CleanupPending (write-ahead) +//! → completion (delete|archive) → PROVE it landed → persist Completed +//! └─ error → record cleanup attempt + backoff → CleanupFailed → retry | give up //! //! if ANY destination permanently exhausts its retry budget: //! the item can never complete (even though other destinations already succeeded) → @@ -18,11 +20,21 @@ //! //! The **write-ahead** ordering is load-bearing (DESIGN §13.2/§20-B): each destination's //! `DestPhase::Verified` is persisted *before* the aggregate item is promoted to `ItemState::Verified`, -//! which itself is persisted *before* the source side effect, and `Completed` *after* — so a crash at -//! any point is recovered idempotently by [`recover`](Worker::recover): an already-`Verified` -//! destination is never re-delivered (the destination object already matches — stable key → idempotent -//! overwrite), and the completion action (delete/archive) fires **exactly once**, only after the last -//! destination verifies. +//! which itself is persisted *before* `ItemState::CleanupPending`, which is persisted *before* the +//! source side effect — so a crash at any point is recovered idempotently by +//! [`recover`](Worker::recover): an already-`Verified` destination is never re-delivered (the +//! destination object already matches — stable key → idempotent overwrite), and the completion action +//! (delete/archive) fires **exactly once**, only after the last destination verifies. +//! +//! ## Completion is proven, not assumed (DESIGN §20-I) +//! `Completed` means the source was **verifiably** released: the archive target exists at the expected +//! size (and re-hashes to the delivered checksum under `completion.verify = checksum`), or the deleted +//! source is really gone. A failed archive move, an unwritable or unconfigured `archiveDir`, or a +//! failed delete is a **cleanup failure**, not a success: the item stays `CleanupFailed`, `FileArchived` +//! /`FileDeleted` are not emitted, and the `replicated` statistic does not move. Cleanup retries run on +//! their own bounded budget ([`Worker::decide_cleanup`]), independent of the transfer's time-based +//! `giveUpAfter`; when it is spent the item stays `CleanupFailed` with a `FileCleanupFailed` event, +//! visible in `get-status` and re-drivable with the `trigger` command. //! //! Failures are classified by [`ReplError`](crate::error::ReplError): *permanent* errors fail fast to //! that destination's `Exhausted`; *transient*/*integrity* errors back off with **full-jitter @@ -48,6 +60,7 @@ use crate::domain::{ }; use crate::error::{ReplError, Result}; use crate::events::{Event, Events, ProgressThrottle}; +use crate::integrity::{hash_reader, verify_checksum, verify_size, Algorithm}; use crate::metrics::{MetricValues, ReplicatorMetrics}; use crate::permission::{PermissionLog, Role}; use crate::ratelimit::Bandwidth; @@ -64,6 +77,18 @@ const DEFAULT_BASE_DELAY_MS: u64 = 1_000; const DEFAULT_MAX_DELAY_MS: u64 = 900_000; const DEFAULT_GIVE_UP_AFTER_MS: i64 = 7 * 24 * 60 * 60 * 1_000; +/// Cleanup attempt budget when `retry.maxAttempts` is not configured (DESIGN §20-I). The source +/// completion action is a local filesystem operation, so it is bounded by **attempts** rather than by +/// the transfer's time-based `giveUpAfter` — that clock starts at discovery and is often largely spent +/// by the time the transfer finishes, which would leave a slow file with no cleanup retries at all. +/// Ten attempts on the shared full-jitter backoff (capped at `retry.maxDelayMs`, default 15 min) spans +/// roughly an hour of transient unavailability before an operator is asked to intervene. +const DEFAULT_CLEANUP_MAX_ATTEMPTS: u32 = 10; + +/// The `next_attempt_at` written for a `CleanupFailed` item whose cleanup budget is spent: no +/// reconciliation scan will ever re-drive it, only an explicit operator `trigger` (DESIGN §20-I). +const CLEANUP_NO_RETRY: i64 = i64::MAX; + /// The resolved retry/backoff policy for an instance (instance `retry` ▸ `global.defaults.retry` ▸ /// built-in defaults, field-by-field). #[derive(Debug, Clone, Copy)] @@ -259,8 +284,8 @@ pub struct Worker { /// Optional metric emitter. metrics: Option>, /// UNS event emitter for the per-file lifecycle events (`ReplicationStarted`/`…Progress`/ - /// `…Completed`/`…Failed`/`FileDeleted`/`FileArchived`/`RetriesExhausted`/`FileQuarantined`, - /// DESIGN §17). A no-op [`Events::disabled`] by default, so a worker built without messaging (or + /// `…Completed`/`…Failed`/`FileDeleted`/`FileArchived`/`FileCleanupFailed`/`RetriesExhausted`/ + /// `FileQuarantined`, DESIGN §17). A no-op [`Events::disabled`] by default, so a worker built without messaging (or /// in a unit test) runs the exact P1/P2 pipeline with zero event overhead. events: Events, /// Feature A (`src/permission.rs`) dedup-log for egress permission errors — ALWAYS logged once @@ -273,6 +298,10 @@ pub struct Worker { /// (ingress evicts recovered paths every tick; egress is inherently bounded by destination count), /// so they are deliberately NOT shared. A fresh, unshared [`PermissionLog`] by default. perm_log: Arc, + /// The source-side filesystem the completion action runs against. Always the real filesystem in + /// production ([`SourceFs::real`]); tests swap in a faulting handle to exercise the cleanup-failure + /// paths (DESIGN §20-I). + source_fs: SourceFs, } impl Worker { @@ -335,9 +364,19 @@ impl Worker { metrics, events: Events::disabled(), perm_log: Arc::new(PermissionLog::new()), + source_fs: SourceFs::real(), } } + /// Swap in a faulting source filesystem (tests only — see [`SourceFs`]). + #[cfg(test)] + fn with_source_faults(mut self, faults: SourceFaults) -> Self { + self.source_fs = SourceFs { + faults: Some(Arc::new(faults)), + }; + self + } + /// Attach the UNS event emitter (the P3 control-plane wiring path, [`crate::app`]). Consumes and /// returns `self` so it composes in [`crate::instance::Instance::build_with_dest`] before the /// worker is shared behind an `Arc`. With the default [`Events::disabled`] every emit is a no-op. @@ -379,8 +418,9 @@ impl Worker { join_rel(&self.ingress_root, &item.relpath) } - /// Process one claimed (`InProgress`) item to a terminal (or `Failed` retry) state, emitting - /// metrics. A durable-store fault is logged and the item is left as-is for the next recovery pass. + /// Process one claimed (`InProgress`) item to a terminal state, a `Failed` transfer retry, or a + /// `CleanupFailed` completion retry (DESIGN §20-I), emitting metrics. A durable-store fault is + /// logged and the item is left as-is for the next recovery pass. pub async fn process_item(&self, item: &WorkItem, now: i64) -> ItemState { // The store returns `relpath` only; rebuild the live absolute source path for delivery. let mut item = item.clone(); @@ -651,42 +691,103 @@ impl Worker { .clear_resume(&self.instance, &item.relpath, label); } - /// Run the success completion action (`delete` | `archive`) on the source, then persist - /// `Completed`. Idempotent: a missing source (already completed before a crash) is fine. Filesystem - /// errors are logged but still advance to `Completed` — the delivery succeeded and the durable - /// `Completed` marker stops the file being re-discovered, so a stale source can never loop. + /// Drive an item whose every destination is `Verified` through the source completion action + /// (`delete` | `archive`) to `Completed` — or, if that action fails, to `CleanupFailed` + /// (DESIGN §20-I). /// - /// The source-side filesystem work runs on the blocking pool: a cross-device archive falls back to - /// a whole-file copy, which must not block a shared async worker thread (DESIGN §6.3). + /// Thin wrapper over [`run_cleanup`](Self::run_cleanup), kept as the name the aggregate-`Verified` + /// call sites use ([`run_pipeline`](Self::run_pipeline), [`recover_verified`](Self::recover_verified), + /// and the window-close pause path). async fn complete_verified(&self, item: &WorkItem, now: i64) -> Result { - let src = self.abs_source(item); - let on_success = self.completion.on_success; - let archive_dir = self.completion.archive_dir.clone(); - let collision = self.completion.on_collision; - let relpath = item.relpath.clone(); - let instance = self.instance.clone(); - if let Err(e) = tokio::task::spawn_blocking(move || { - apply_success_action( - on_success, - &src, - archive_dir.as_deref(), - &relpath, - collision, - &instance, - ) - }) - .await - { - tracing::warn!( - instance = %self.instance, relpath = %item.relpath, error = %e, - "completion task join failed; marking complete anyway (delivery already succeeded)" - ); + self.run_cleanup(item, now).await + } + + /// Apply the source completion action and persist the outcome (DESIGN §13.2/§20-I). + /// + /// **Write-ahead:** `CleanupPending` is persisted *before* the filesystem is touched, so a crash + /// mid-action is found by [`recover`](Self::recover) and re-evaluated against observed state. + /// `Completed` is persisted only once [`apply_success_action`] has **proven** the action landed — + /// the archive target exists at the source's size (re-hashed to the delivered checksum under + /// `completion.verify = checksum`), or the deleted source is really gone. Anything else is a + /// cleanup failure: the item moves to `CleanupFailed`, no `FileArchived`/`FileDeleted` is emitted, + /// and the `replicated` statistic does not move. For an evidence pipeline a source left behind, or + /// an archive copy that never landed, is not a success and must never be recorded as one. + /// + /// The filesystem work runs on the blocking pool: a cross-device archive falls back to a whole-file + /// copy plus a re-hash, which must not block a shared async worker thread (DESIGN §6.3). + /// + /// Callers own the metric emission ([`emit`](Self::emit)) for the returned state. + pub(crate) async fn run_cleanup(&self, item: &WorkItem, now: i64) -> Result { + self.store.set_state( + &self.instance, + &item.relpath, + ItemState::CleanupPending, + now, + )?; + + let ctx = CleanupCtx { + on_success: self.completion.on_success, + src: self.abs_source(item), + archive_dir: self.completion.archive_dir.clone(), + relpath: item.relpath.clone(), + collision: self.completion.on_collision, + verify: self.completion.verify, + expected_size: item.size, + expected_checksum: self.delivered_checksum(&item.relpath), + }; + let action = ctx.action(); + let fs = self.source_fs.clone(); + let outcome = + match tokio::task::spawn_blocking(move || apply_success_action(&ctx, &fs)).await { + Ok(r) => r, + Err(join_err) => Err(ReplError::Transient(format!( + "completion task join failed: {join_err}" + ))), + }; + match outcome { + Ok(done) => self.finish_completed(item, done, now).await, + Err(e) => self.record_cleanup_failure(item, action, e, now).await, + } + } + + /// The checksum every destination verified this item's bytes against, read back from any + /// destination's `Verified` write-ahead checkpoint (they all hashed the same source bytes, and the + /// [`Checksum`] variant carries its own algorithm). The checkpoints are cleared only AFTER + /// `Completed`, so they are still present for every cleanup attempt, including one that runs after + /// a restart. [`Checksum::None`] when nothing was hashed on delivery. + fn delivered_checksum(&self, relpath: &str) -> Checksum { + for slot in &self.dests { + let Ok(Some(resume)) = self.store.load_resume(&self.instance, relpath, &slot.label) + else { + continue; + }; + let delivered: Option = resume + .token + .get("verified") + .cloned() + .and_then(|v| serde_json::from_value(v).ok()); + if let Some(d) = delivered { + if d.checksum != Checksum::None { + return d.checksum; + } + } } + Checksum::None + } + + /// Persist `Completed` after the source action was proven done, then emit the source-side success + /// event and count the file as replicated. + async fn finish_completed( + &self, + item: &WorkItem, + done: CleanupDone, + now: i64, + ) -> Result { self.store .set_state(&self.instance, &item.relpath, ItemState::Completed, now)?; tracing::info!( instance = %self.instance, relpath = %item.relpath, - action = ?self.completion.on_success, "item completed; source side-effect applied" + action = ?self.completion.on_success, "item completed; source side-effect verified" ); // Clear every destination's resume checkpoint + per-destination completion bookkeeping only // AFTER Completed is durable, so a crash in the aggregate Verified→Completed gap still finds @@ -702,27 +803,24 @@ impl Worker { }, )?; - // Lifecycle events (DESIGN §17.1): the source side effect that ran. `ReplicationCompleted` was - // already emitted per destination (by `run_one_dest`, or by `recover_verified` re-verifying on - // recovery) — this is the aggregate/source-side event, fired exactly once. - match self.completion.on_success { - OnSuccess::Delete => { + // Lifecycle events (DESIGN §17.1): the source side effect that actually ran. + // `ReplicationCompleted` was already emitted per destination (by `run_one_dest`, or by + // `recover_verified` re-verifying on recovery) — this is the aggregate/source-side event, fired + // exactly once, and only for an action that verifiably happened. `archivePath` is the path the + // file really landed at (which the `suffix` collision policy can rename), not a computed guess. + match done { + CleanupDone::Deleted => { self.events .emit(Event::FileDeleted { path: item.relpath.clone(), }) .await; } - OnSuccess::Archive => { - let archive_path = self - .completion - .archive_dir - .as_ref() - .map(|d| join_rel(d, &item.relpath).display().to_string()); + CleanupDone::Archived(target) => { self.events .emit(Event::FileArchived { path: item.relpath.clone(), - archive_path, + archive_path: target.map(|p| p.display().to_string()), }) .await; } @@ -730,6 +828,125 @@ impl Worker { Ok(ItemState::Completed) } + /// Record a failed source completion action (DESIGN §20-I): `cleanup_attempts += 1`, the error, and + /// the move to `CleanupFailed` with the item's next cleanup gate. While the cleanup budget lasts the + /// gate is a full-jitter backoff and the next reconciliation scan re-drives the item; once the + /// budget is spent the gate is [`CLEANUP_NO_RETRY`] (only an operator `trigger` re-drives it) and a + /// `FileCleanupFailed` event is emitted. + async fn record_cleanup_failure( + &self, + item: &WorkItem, + action: &str, + error: ReplError, + now: i64, + ) -> Result { + let attempts = item.cleanup_attempts.saturating_add(1); + let err_str = error.to_string(); + match self.decide_cleanup(error.is_permanent(), item.cleanup_attempts, now) { + RetryDecision::Retry { next_attempt_at } => { + self.store.record_cleanup_attempt( + &self.instance, + &item.relpath, + &err_str, + ItemState::CleanupFailed, + next_attempt_at, + now, + )?; + tracing::warn!( + instance = %self.instance, relpath = %item.relpath, action, + attempts, retry_at = next_attempt_at, error = %err_str, + "source completion action failed; scheduled for retry (item NOT completed)" + ); + } + RetryDecision::GiveUp => { + self.store.record_cleanup_attempt( + &self.instance, + &item.relpath, + &err_str, + ItemState::CleanupFailed, + CLEANUP_NO_RETRY, + now, + )?; + tracing::error!( + instance = %self.instance, relpath = %item.relpath, action, attempts, + error = %err_str, + "source completion action failed permanently; item parked in CleanupFailed \ + (delivered and verified on every destination, source NOT released)" + ); + self.events + .emit(Event::FileCleanupFailed { + path: item.relpath.clone(), + action: action.to_string(), + attempts, + last_error: err_str, + }) + .await; + } + } + Ok(ItemState::CleanupFailed) + } + + /// Decide whether a failed cleanup attempt is retried (DESIGN §20-I). Uses the instance's + /// [`RetryPolicy`] backoff — the same full-jitter exponential curve, capped at `retry.maxDelayMs` — + /// but its OWN attempt budget: `retry.maxAttempts` when configured, otherwise + /// [`DEFAULT_CLEANUP_MAX_ATTEMPTS`]. + /// + /// Deliberately NOT the transfer's time-based `giveUpAfter`: that clock starts at discovery and is + /// usually largely spent by the time a slow or long-retried transfer finishes, which would leave + /// exactly the files that struggled hardest with no cleanup retries at all. A permanent error — an + /// unconfigured `archiveDir`, an archived copy that does not match — gives up immediately, because + /// retrying cannot change either and the operator needs it surfaced. A **locked** source (another + /// process holding it open) is deliberately transient here and retried, unlike on the transfer path: + /// see [`ReplError::classify_cleanup_io`]. + fn decide_cleanup(&self, permanent: bool, attempts_so_far: u32, now: i64) -> RetryDecision { + if permanent { + return RetryDecision::GiveUp; + } + let attempts = attempts_so_far.saturating_add(1); + let cap = self + .retry + .max_attempts + .unwrap_or(DEFAULT_CLEANUP_MAX_ATTEMPTS); + if attempts >= cap { + return RetryDecision::GiveUp; + } + let delay = { + let mut r = self.rng.lock().expect("rng mutex"); + self.retry.backoff_ms(attempts, &mut *r) + } as i64; + RetryDecision::Retry { + next_attempt_at: now.saturating_add(delay), + } + } + + /// Re-drive the items whose delivery is done but whose source completion action is not + /// (DESIGN §20-I) — every `CleanupPending` row (a crash or an abort left the action mid-flight) + /// plus the `CleanupFailed` rows whose cleanup backoff gate has elapsed. + /// + /// Called on every reconciliation tick, so a transient cleanup failure heals on its own. With + /// `redrive_all` (the operator `trigger` command) the gate is ignored, which is what rescues an + /// item whose cleanup budget is spent once the underlying cause — a full disk, a missing + /// `archiveDir`, a read-only mount — has been fixed. + pub async fn drive_cleanup(&self, now: i64, redrive_all: bool) -> Result<()> { + let mut due = self + .store + .list_by_state(&self.instance, ItemState::CleanupPending)?; + for it in self + .store + .list_by_state(&self.instance, ItemState::CleanupFailed)? + { + if redrive_all || it.next_attempt_at <= now { + due.push(it); + } + } + for mut it in due { + it.abs_source = self.abs_source(&it); + let state = self.run_cleanup(&it, now).await?; + self.emit(state, it.size).await; + } + Ok(()) + } + /// Terminal handling for an `Exhausted` item: `quarantine` (move to `failedDir` + an /// `.error.json` sidecar) or `retainInPlace` (leave it, mark `Retained`). Idempotent for recovery. /// The quarantine move + sidecar write run on the blocking pool (cross-device moves copy). @@ -770,8 +987,9 @@ impl Worker { destination: self.dest_labels_joined(), bytes_done: item.bytes_done, }; + let fs = self.source_fs.clone(); if let Err(e) = tokio::task::spawn_blocking(move || { - apply_quarantine_action(failed_dir.as_deref(), &src, collision, &ctx) + apply_quarantine_action(failed_dir.as_deref(), &src, collision, &ctx, &fs) }) .await { @@ -810,14 +1028,30 @@ impl Worker { } } - /// Crash recovery (DESIGN §13.2), idempotent. `Verified` items are **re-verified against the - /// destination before** the source side effect (see [`recover_verified`](Self::recover_verified)); - /// `InProgress` items return to `Ready` for idempotent re-delivery; `Exhausted` items (a crash - /// before their terminal action) re-run `onExhausted`. + /// Crash recovery (DESIGN §13.2), idempotent, run before any new work. `Verified` items are + /// **re-verified against the destination before** the source side effect (see + /// [`recover_verified`](Self::recover_verified)); `CleanupPending` items — a crash caught between + /// the write-ahead cleanup marker and the proof that the action landed — are re-evaluated against + /// observed filesystem state by [`run_cleanup`](Self::run_cleanup) (DESIGN §20-I): a source still + /// present means the action is retried, a source already gone (deleted, or archived — `move_file` + /// removes it only after the target rename succeeds) means the item completes. `InProgress` items + /// return to `Ready` for idempotent re-delivery; `Exhausted` items (a crash before their terminal + /// action) re-run `onExhausted`. + /// + /// `CleanupFailed` items are NOT re-driven here: they carry their own backoff gate and are picked + /// up by [`drive_cleanup`](Self::drive_cleanup) on the reconciliation tick that follows startup. pub async fn recover(&self, now: i64) -> Result<()> { for it in self.store.recover_incomplete(&self.instance)? { match it.state { ItemState::Verified => self.recover_verified(&it, now).await?, + ItemState::CleanupPending => { + tracing::info!( + instance = %self.instance, relpath = %it.relpath, + "recovering CleanupPending → re-evaluating the source completion action" + ); + let state = self.run_cleanup(&it, now).await?; + self.emit(state, it.size).await; + } ItemState::InProgress => { // Before re-readying, re-verify any destination a PRIOR session already marked // `DestPhase::Verified` (fan-out narrows the §13.2 crash window — see @@ -883,8 +1117,10 @@ impl Worker { if all_ok { tracing::info!(instance = %self.instance, relpath = %it.relpath, "recovering Verified → every destination re-verified, completing"); - let _ = self.complete_verified(it, now).await?; - self.emit(ItemState::Completed, it.size).await; + // The completion action can still fail (an unwritable archive, a vanished archiveDir), so + // report the state it actually reached — `Completed` or `CleanupFailed` (DESIGN §20-I). + let state = self.complete_verified(it, now).await?; + self.emit(state, it.size).await; } else { self.store .set_state(&self.instance, &it.relpath, ItemState::Ready, now)?; @@ -1547,6 +1783,124 @@ fn percent_int(done: u64, size: u64) -> i32 { (((done as f64 / size as f64) * 100.0).round() as i32).clamp(0, 100) } +/// The source-side filesystem operations the completion action performs, behind one cloneable handle +/// so tests can inject faults (a failing rename, an uncreatable archive directory, a failing delete, a +/// cross-filesystem copy) deterministically — no real cross-device mount and no platform-specific +/// permission tricks. In production every call delegates straight to `std::fs`; the fault table exists +/// only in test builds, mirroring [`Events`]'s test recorder, so the shipping path carries nothing but +/// the plain `std::fs` call. +#[derive(Clone, Default)] +pub(crate) struct SourceFs { + #[cfg(test)] + faults: Option>, +} + +impl SourceFs { + /// The real filesystem (the only constructor used outside tests). + pub(crate) fn real() -> Self { + Self::default() + } + + /// The injected error for `op`, if this handle is faulting it on this call. + #[cfg(test)] + fn fault(&self, op: FsOp) -> Option { + self.faults.as_ref().and_then(|f| f.take(op)) + } + #[cfg(not(test))] + #[inline] + fn fault(&self, _op: FsOp) -> Option { + None + } + + fn remove_file(&self, path: &Path) -> std::io::Result<()> { + match self.fault(FsOp::RemoveFile) { + Some(e) => Err(e), + None => std::fs::remove_file(path), + } + } + + fn create_dir_all(&self, path: &Path) -> std::io::Result<()> { + match self.fault(FsOp::CreateDirAll) { + Some(e) => Err(e), + None => std::fs::create_dir_all(path), + } + } + + fn rename(&self, from: &Path, to: &Path) -> std::io::Result<()> { + match self.fault(FsOp::Rename) { + Some(e) => Err(e), + None => std::fs::rename(from, to), + } + } + + fn copy(&self, from: &Path, to: &Path) -> std::io::Result { + match self.fault(FsOp::Copy) { + Some(e) => Err(e), + None => std::fs::copy(from, to), + } + } +} + +/// How a filesystem `io::Error` is turned into a [`ReplError`]. Passed explicitly so the shared +/// [`move_file`] helper keeps each caller's own policy: the source completion action classifies a +/// locked file as transient ([`ReplError::classify_cleanup_io`]), while quarantine keeps the ordinary +/// [`ReplError::classify_io`] rules (DESIGN §20-I). +type IoClassifier = fn(std::io::Error) -> ReplError; + +/// The [`SourceFs`] operations a test can fault. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum FsOp { + RemoveFile, + CreateDirAll, + Rename, + Copy, +} + +/// Test-only fault table for [`SourceFs`]: each faulted operation returns `kind` instead of touching +/// the filesystem. `heal_after` bounds how many faulted calls are served before the table stops +/// faulting, which is what lets a test prove "cleanup retries, then succeeds". +#[cfg(test)] +#[derive(Debug, Default)] +pub(crate) struct SourceFaults { + ops: Vec<(FsOp, std::io::ErrorKind)>, + heal_after: Option, + calls: std::sync::atomic::AtomicU32, +} + +#[cfg(test)] +impl SourceFaults { + /// Fault `op` with `kind` on every call. + pub(crate) fn always(op: FsOp, kind: std::io::ErrorKind) -> Self { + SourceFaults { + ops: vec![(op, kind)], + heal_after: None, + calls: std::sync::atomic::AtomicU32::new(0), + } + } + + /// Also fault `op` with `kind` (chained onto an existing table). + pub(crate) fn and(mut self, op: FsOp, kind: std::io::ErrorKind) -> Self { + self.ops.push((op, kind)); + self + } + + /// Stop faulting after `n` faulted calls have been served. + pub(crate) fn heal_after(mut self, n: u32) -> Self { + self.heal_after = Some(n); + self + } + + fn take(&self, op: FsOp) -> Option { + let kind = self.ops.iter().find(|(o, _)| *o == op).map(|(_, k)| *k)?; + if let Some(limit) = self.heal_after { + if self.calls.fetch_add(1, Ordering::SeqCst) >= limit { + return None; + } + } + Some(std::io::Error::new(kind, format!("injected {op:?} fault"))) + } +} + /// Join a forward-slash `relpath` onto `root`, dropping empty/`.`/`..` segments so the result always /// stays under `root` (mirrors the destination key derivation). fn join_rel(root: &Path, relpath: &str) -> PathBuf { @@ -1560,45 +1914,55 @@ fn join_rel(root: &Path, relpath: &str) -> PathBuf { p } -/// Move `src` → `dst`, creating parents and honoring the collision policy. Prefers an atomic rename; -/// falls back to a **crash-atomic** copy across filesystems. +/// Move `src` → `dst`, creating parents and honoring the collision policy, returning the **resolved** +/// target the file actually landed at (which differs from `dst` under the `suffix` policy). Prefers an +/// atomic rename; falls back to a **crash-atomic** copy across filesystems. /// /// The cross-device fallback copies into a sibling temp file in the destination directory, then /// atomically renames it onto the resolved target and removes the source. Copying via a temp keeps /// the move crash-atomic: a crash *during* the (long) copy leaves only an orphan temp, never a /// partially-written `target` that a recovery pass would mistake for a real file and duplicate under /// the default `suffix` collision policy (`name` + `name.1.ext`). (DESIGN §13.2.) -fn move_file(src: &Path, dst: &Path, collision: Collision) -> Result<()> { +/// +/// Every filesystem call goes through [`SourceFs`] so the failure branches are testable (DESIGN §20-I); +/// the caller reports the returned path and verifies the file landed there. +fn move_file( + src: &Path, + dst: &Path, + collision: Collision, + fs: &SourceFs, + classify: IoClassifier, +) -> Result { if let Some(parent) = dst.parent() { - std::fs::create_dir_all(parent).map_err(ReplError::classify_io)?; + fs.create_dir_all(parent).map_err(classify)?; } - let target = resolve_collision(dst, collision)?; + let target = resolve_collision(dst, collision, fs, classify)?; // Fast path: same-filesystem rename is already atomic. - if std::fs::rename(src, &target).is_ok() { - return Ok(()); + if fs.rename(src, &target).is_ok() { + return Ok(target); } // Cross-device (or Windows target-exists) fallback: copy → temp, atomic rename, remove source. let tmp = move_temp_path(&target); - if let Err(e) = std::fs::copy(src, &tmp) { + if let Err(e) = fs.copy(src, &tmp) { let _ = std::fs::remove_file(&tmp); - return Err(ReplError::classify_io(e)); + return Err(classify(e)); } - if std::fs::rename(&tmp, &target).is_err() { + if fs.rename(&tmp, &target).is_err() { // Windows rename won't overwrite; `resolve_collision` should have freed the target, but be // defensive: remove an existing target then retry, cleaning the temp on any failure. if target.exists() { - if let Err(e) = std::fs::remove_file(&target) { + if let Err(e) = fs.remove_file(&target) { let _ = std::fs::remove_file(&tmp); - return Err(ReplError::classify_io(e)); + return Err(classify(e)); } } - if let Err(e) = std::fs::rename(&tmp, &target) { + if let Err(e) = fs.rename(&tmp, &target) { let _ = std::fs::remove_file(&tmp); - return Err(ReplError::classify_io(e)); + return Err(classify(e)); } } - std::fs::remove_file(src).map_err(ReplError::classify_io)?; - Ok(()) + fs.remove_file(src).map_err(classify)?; + Ok(target) } /// A collision-resistant temp path in the destination directory for the crash-atomic move fallback: @@ -1616,42 +1980,143 @@ fn move_temp_path(target: &Path) -> PathBuf { dir.join(format!(".{name}.{tag:032x}.movetmp")) } -/// The success completion side effect (`delete` | `archive`) on the source file, run on the blocking -/// pool by [`Worker::complete_verified`]. Idempotent: a missing source (already completed before a -/// crash) is fine. Errors are logged, not fatal — delivery already succeeded and the durable -/// `Completed` marker stops re-discovery, so a stale source can never loop. -fn apply_success_action( +/// Everything the source-side completion action needs, carried into the blocking pool by +/// [`Worker::run_cleanup`]. +struct CleanupCtx { on_success: OnSuccess, - src: &Path, - archive_dir: Option<&Path>, - relpath: &str, + /// Absolute source path. + src: PathBuf, + archive_dir: Option, + relpath: String, collision: Collision, - instance: &str, -) { - match on_success { + /// `completion.verify` — decides whether the archived copy is re-hashed or only size-checked. + verify: Verify, + /// The source's byte count, as every destination verified it. + expected_size: u64, + /// The checksum the destinations verified against, recovered from a `Verified` write-ahead + /// checkpoint. [`Checksum::None`] when nothing was hashed (`completion.verify` is `size`/`none`, + /// or the checkpoint is gone), which falls the archive proof back to the size check. + expected_checksum: Checksum, +} + +impl CleanupCtx { + /// The wire token for the action this context performs (`FileCleanupFailed.action`). + fn action(&self) -> &'static str { + match self.on_success { + OnSuccess::Delete => "delete", + OnSuccess::Archive => "archive", + } + } +} + +/// What the completion action verifiably did — the input to the success event. +enum CleanupDone { + /// The source is gone. + Deleted, + /// The source landed in the archive at this path. `None` only when a prior, crashed attempt had + /// already moved it under a collision-resolved name this pass cannot re-derive. + Archived(Option), +} + +/// The success completion side effect (`delete` | `archive`) on the source file, run on the blocking +/// pool by [`Worker::run_cleanup`], returning **only** once the action is proven to have happened +/// (DESIGN §20-I). +/// +/// Idempotent for recovery: a source a prior attempt already released is re-checked against observed +/// filesystem state rather than re-applied — an absent source under `delete` is done, and an absent +/// source under `archive` means the move's final rename already succeeded (`move_file` removes the +/// source only afterwards), so the archived copy is re-verified where it can still be identified. +/// +/// Every failure is returned, never swallowed: an unwritable or unconfigured `archiveDir`, a failing +/// rename or copy, a failing delete, and a target that does not match the source are all cleanup +/// failures that keep the item out of `Completed`. +fn apply_success_action(ctx: &CleanupCtx, fs: &SourceFs) -> Result { + match ctx.on_success { OnSuccess::Delete => { - if let Err(e) = std::fs::remove_file(src) { - if e.kind() != std::io::ErrorKind::NotFound { - tracing::warn!(path = %src.display(), error = %e, "delete source failed"); - } + match fs.remove_file(&ctx.src) { + Ok(()) => {} + // Already gone: a prior attempt (or one interrupted by a crash) succeeded. + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(ReplError::classify_cleanup_io(e)), + } + if ctx.src.exists() { + return Err(ReplError::Transient(format!( + "delete returned success but the source is still present: {}", + ctx.src.display() + ))); } + Ok(CleanupDone::Deleted) } - OnSuccess::Archive => match archive_dir { - Some(dir) if src.exists() => { - let dst = join_rel(dir, relpath); - if let Err(e) = move_file(src, &dst, collision) { - tracing::warn!( - src = %src.display(), dst = %dst.display(), error = %e, - "archive move failed; leaving source in place" - ); + OnSuccess::Archive => { + // A missing `archiveDir` cannot archive anything. Permanent: no number of retries makes a + // configuration appear, and the operator needs it surfaced now rather than in an hour. + let Some(dir) = ctx.archive_dir.as_deref() else { + return Err(ReplError::Permanent(format!( + "onSuccess=archive requires completion.archiveDir; cannot archive {}", + ctx.relpath + ))); + }; + let dst = join_rel(dir, &ctx.relpath); + if !ctx.src.exists() { + // The source was already released by a prior attempt. If the un-suffixed target is + // there, prove it; otherwise the file landed under a collision-resolved name, and with + // the source gone and every destination verified the item is complete. + if dst.exists() { + verify_archived(&dst, ctx)?; + return Ok(CleanupDone::Archived(Some(dst))); } + tracing::warn!( + relpath = %ctx.relpath, dst = %dst.display(), + "source already archived under a collision-resolved name; completing without a path" + ); + return Ok(CleanupDone::Archived(None)); } - Some(_) => {} // source already gone (idempotent recovery) - None => tracing::warn!( - instance = %instance, relpath = %relpath, - "onSuccess=archive but no archiveDir configured; leaving source in place" - ), - }, + let target = move_file( + &ctx.src, + &dst, + ctx.collision, + fs, + ReplError::classify_cleanup_io, + )?; + verify_archived(&target, ctx)?; + Ok(CleanupDone::Archived(Some(target))) + } + } +} + +/// Prove an archived file is really there and really the file (DESIGN §20-I): it must exist at the +/// source's byte count and, under `completion.verify = checksum`, re-hash to the checksum every +/// destination verified against. +/// +/// A failure to *read* the target (an unmounted archive volume, a transient I/O error) is classified +/// as usual and retried. A genuine size/checksum **mismatch** is returned as +/// [`ReplError::Permanent`]: the move already released the source, so no retry can change the outcome +/// and the operator needs the item parked in `CleanupFailed` with the reason immediately. +fn verify_archived(target: &Path, ctx: &CleanupCtx) -> Result<()> { + let len = std::fs::metadata(target) + .map_err(ReplError::classify_cleanup_io)? + .len(); + verify_size(ctx.expected_size, len).map_err(permanent_mismatch)?; + if ctx.verify != Verify::Checksum { + return Ok(()); + } + let algo = match ctx.expected_checksum { + Checksum::Crc32c(_) => Algorithm::Crc32c, + Checksum::Sha256(_) => Algorithm::Sha256, + // Nothing was hashed on delivery, so the size check above is the whole proof available. + Checksum::None => return Ok(()), + }; + let mut f = std::fs::File::open(target).map_err(ReplError::classify_cleanup_io)?; + let (_, actual) = hash_reader(&mut f, algo).map_err(ReplError::classify_cleanup_io)?; + verify_checksum(&ctx.expected_checksum, &actual).map_err(permanent_mismatch) +} + +/// Re-tag an integrity mismatch on an already-moved archive copy as permanent (see +/// [`verify_archived`]); any other error keeps its classification. +fn permanent_mismatch(e: ReplError) -> ReplError { + match e { + ReplError::Integrity(m) => ReplError::Permanent(format!("archived copy mismatch: {m}")), + other => other, } } @@ -1675,12 +2140,13 @@ fn apply_quarantine_action( src: &Path, collision: Collision, ctx: &QuarantineCtx, + fs: &SourceFs, ) { match failed_dir { Some(dir) => { let dst = join_rel(dir, &ctx.relpath); if src.exists() { - if let Err(e) = move_file(src, &dst, collision) { + if let Err(e) = move_file(src, &dst, collision, fs, ReplError::classify_io) { tracing::warn!( src = %src.display(), dst = %dst.display(), error = %e, "quarantine move failed" @@ -1724,13 +2190,18 @@ fn write_error_sidecar(dst: &Path, ctx: &QuarantineCtx) { /// Resolve the effective target path for a collision policy: `overwrite` removes the existing file, /// `suffix` finds a free `name.N.ext`, `fail` errors permanently. -fn resolve_collision(dst: &Path, collision: Collision) -> Result { +fn resolve_collision( + dst: &Path, + collision: Collision, + fs: &SourceFs, + classify: IoClassifier, +) -> Result { if !dst.exists() { return Ok(dst.to_path_buf()); } match collision { Collision::Overwrite => { - std::fs::remove_file(dst).map_err(ReplError::classify_io)?; + fs.remove_file(dst).map_err(classify)?; Ok(dst.to_path_buf()) } Collision::Fail => Err(ReplError::Permanent(format!( @@ -1872,6 +2343,7 @@ mod tests { size: 1, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -1921,7 +2393,14 @@ mod tests { let src = dir.path().join("src.txt"); std::fs::write(&src, b"hi").unwrap(); let dst = dir.path().join("sub/dir/out.txt"); - move_file(&src, &dst, Collision::Fail).unwrap(); + move_file( + &src, + &dst, + Collision::Fail, + &SourceFs::real(), + ReplError::classify_io, + ) + .unwrap(); assert!(!src.exists()); assert_eq!(std::fs::read(&dst).unwrap(), b"hi"); } @@ -1938,18 +2417,39 @@ mod tests { // Fail → error, source untouched. let s1 = mk("s1.txt", b"a"); - assert!(move_file(&s1, &dst, Collision::Fail).is_err()); + assert!(move_file( + &s1, + &dst, + Collision::Fail, + &SourceFs::real(), + ReplError::classify_io + ) + .is_err()); assert!(s1.exists()); // Suffix → writes dst.1.txt, original preserved. let s2 = mk("s2.txt", b"b"); - move_file(&s2, &dst, Collision::Suffix).unwrap(); + move_file( + &s2, + &dst, + Collision::Suffix, + &SourceFs::real(), + ReplError::classify_io, + ) + .unwrap(); assert_eq!(std::fs::read(dir.path().join("dst.1.txt")).unwrap(), b"b"); assert_eq!(std::fs::read(&dst).unwrap(), b"existing"); // Overwrite → replaces dst. let s3 = mk("s3.txt", b"c"); - move_file(&s3, &dst, Collision::Overwrite).unwrap(); + move_file( + &s3, + &dst, + Collision::Overwrite, + &SourceFs::real(), + ReplError::classify_io, + ) + .unwrap(); assert_eq!(std::fs::read(&dst).unwrap(), b"c"); } @@ -2383,6 +2883,7 @@ mod tests { size: 18, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -2458,6 +2959,7 @@ mod tests { size: 16, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -3409,6 +3911,7 @@ mod tests { size: 7, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -3631,6 +4134,7 @@ mod tests { size: 7, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -3764,6 +4268,7 @@ mod tests { size: 7, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, @@ -3888,4 +4393,780 @@ mod tests { "completes exactly once" ); } + + // ---- source-completion failures are NOT success (DESIGN §20-I) ------------------------------- + // + // Every test below shares one invariant: a file that replicated and verified on every destination + // but whose source completion action did NOT verifiably happen must never reach `Completed`, never + // emit `FileArchived`/`FileDeleted`, and never move the `replicated` statistic. For an evidence + // pipeline that combination is a false record of custody. + + /// A worker with an archive completion pointing at `archive_root`, and an injected source-side + /// filesystem fault table. + fn archiving_worker( + store: Arc, + src_root: &Path, + dst_root: &Path, + archive_root: Option<&Path>, + retry: RetryPolicy, + faults: SourceFaults, + ) -> Worker { + let mut comp = completion(OnSuccess::Archive); + comp.archive_dir = archive_root.map(|p| p.to_path_buf()); + local_worker(store, src_root, dst_root, comp, retry).with_source_faults(faults) + } + + /// Claim the single ready item and run it through the whole pipeline. + async fn process_only_item( + worker: &Worker, + store: &Arc, + now: i64, + ) -> ItemState { + let item = store.claim_ready(INST, 10, now).unwrap().pop().unwrap(); + worker.process_item(&item, now).await + } + + #[tokio::test] + async fn archive_move_failure_leaves_the_item_cleanup_failed_not_completed() { + // The defect this fixes: the transfer succeeded, the archive move did not, and the item was + // still recorded as `Completed` + `FileArchived` with a computed path that pointed at nothing. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + // Both the rename fast path and the cross-device copy fallback fail: the move cannot happen. + let worker = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + RetryPolicy::default(), + SourceFaults::always(FsOp::Rename, std::io::ErrorKind::Other) + .and(FsOp::Copy, std::io::ErrorKind::Other), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.state, ItemState::CleanupFailed, "never Completed"); + assert_eq!(row.cleanup_attempts, 1); + assert!(row.last_error.is_some()); + assert!( + src.path().join("r.csv").exists(), + "the source is still there — that is exactly why this is not a success" + ); + assert!( + !archive.path().join("r.csv").exists(), + "nothing was archived" + ); + assert!( + fake.events_named("FileArchived").is_empty(), + "no FileArchived for a move that never happened" + ); + assert_eq!( + store.stats(INST).unwrap().replicated, + 0, + "a cleanup failure is not a replicated file" + ); + // The delivery itself did succeed and is still reported per destination. + assert_eq!(fake.events_named("ReplicationCompleted").len(), 1); + assert_eq!( + std::fs::read(dst.path().join("r.csv")).unwrap(), + b"evidence" + ); + } + + #[tokio::test] + async fn missing_archive_dir_is_a_cleanup_failure_not_a_silent_retain() { + // `onSuccess: archive` with no `archiveDir` used to log a warning, leave the source in place, + // and still mark the item Completed. It is a permanent cleanup failure: no retry can conjure + // configuration, so the item is parked and the operator is told once. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = archiving_worker( + store.clone(), + src.path(), + dst.path(), + None, // no archiveDir configured + RetryPolicy::default(), + SourceFaults::default(), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.state, ItemState::CleanupFailed); + assert_eq!( + row.next_attempt_at, CLEANUP_NO_RETRY, + "permanent → parked for the operator, not retried on a timer" + ); + assert!(src.path().join("r.csv").exists(), "source retained"); + assert!(fake.events_named("FileArchived").is_empty()); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + + let failures = fake.events_named("FileCleanupFailed"); + assert_eq!(failures.len(), 1); + assert_eq!(failures[0].body["path"], serde_json::json!("r.csv")); + assert_eq!(failures[0].body["action"], serde_json::json!("archive")); + assert_eq!(failures[0].body["attempts"], serde_json::json!(1)); + assert!(failures[0].body["lastError"] + .as_str() + .unwrap() + .contains("archiveDir")); + } + + #[tokio::test] + async fn uncreatable_archive_dir_is_a_cleanup_failure() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + // Two cleanup attempts, so the retry and the give-up are both observable. + let retry = RetryPolicy { + max_attempts: Some(2), + ..RetryPolicy::default() + }; + let worker = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + retry, + SourceFaults::always(FsOp::CreateDirAll, std::io::ErrorKind::PermissionDenied), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + assert!(src.path().join("r.csv").exists()); + assert!(fake.events_named("FileArchived").is_empty()); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + // On the cleanup path a permission denial is TRANSIENT (the archive volume may be briefly + // unavailable, or something may hold the target), so the first attempt schedules a retry rather + // than parking the file. + assert!(fake.events_named("FileCleanupFailed").is_empty()); + assert!( + store.get(INST, "r.csv").unwrap().unwrap().next_attempt_at > 100, + "a retry gate, not the parked sentinel" + ); + + // Once the cleanup budget is spent it parks and reports, still without ever completing. + worker.drive_cleanup(10_000_000, false).await.unwrap(); + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.state, ItemState::CleanupFailed); + assert_eq!(row.next_attempt_at, CLEANUP_NO_RETRY); + assert_eq!(fake.events_named("FileCleanupFailed").len(), 1); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + } + + #[tokio::test] + async fn a_locked_source_file_retries_and_completes_once_the_lock_clears() { + // The ordinary Windows case: a producer, antivirus scanner, indexer, or backup agent still + // holds the source open, so the delete fails with a sharing/permission error. On the CLEANUP + // path that is transient — the file must be retried under the cleanup backoff, not parked after + // a single attempt the way an egress permission denial is (DESIGN §20-I). + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + RetryPolicy::default(), + ) + .with_source_faults( + SourceFaults::always(FsOp::RemoveFile, std::io::ErrorKind::PermissionDenied) + .heal_after(1), + ) + .with_events(events); + + // Attempt 1: the file is locked → retry scheduled, nothing completed, nothing announced. + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.cleanup_attempts, 1); + assert!( + row.next_attempt_at > 100 && row.next_attempt_at != CLEANUP_NO_RETRY, + "a locked source is retried on the backoff, not parked: {row:?}" + ); + assert!( + row.last_error.as_deref().unwrap().starts_with("transient:"), + "classified transient on the cleanup path, got {:?}", + row.last_error + ); + assert!(fake.events_named("FileCleanupFailed").is_empty()); + assert!(fake.events_named("FileDeleted").is_empty()); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + assert!(src.path().join("r.csv").exists()); + + // Attempt 2, after the lock clears: the source is released and the file completes. + worker.drive_cleanup(10_000_000, false).await.unwrap(); + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::Completed + ); + assert!(!src.path().join("r.csv").exists()); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + assert_eq!(fake.events_named("FileDeleted").len(), 1); + assert!(fake.events_named("FileCleanupFailed").is_empty()); + } + + #[tokio::test] + async fn delete_failure_leaves_the_item_cleanup_failed_not_completed() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + RetryPolicy::default(), + ) + .with_source_faults(SourceFaults::always( + FsOp::RemoveFile, + std::io::ErrorKind::Other, + )) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::CleanupFailed + ); + assert!( + src.path().join("r.csv").exists(), + "the source is still there" + ); + assert!(fake.events_named("FileDeleted").is_empty()); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + } + + #[tokio::test] + async fn cross_filesystem_archive_copies_then_verifies_and_completes() { + // The rename fast path fails (as it does across a device boundary); the copy → temp → rename + // fallback runs, the archived copy is re-hashed against the delivered checksum, and only then + // does the item complete. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + RetryPolicy::default(), + // Only the FIRST rename (the same-filesystem fast path) fails; the fallback's temp→target + // rename then succeeds. + SourceFaults::always(FsOp::Rename, std::io::ErrorKind::Other).heal_after(1), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::Completed + ); + assert!(!src.path().join("r.csv").exists(), "source released"); + assert_eq!( + std::fs::read(archive.path().join("r.csv")).unwrap(), + b"evidence" + ); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + let archived = fake.events_named("FileArchived"); + assert_eq!(archived.len(), 1); + assert_eq!( + archived[0].body["archivePath"].as_str().unwrap(), + archive.path().join("r.csv").display().to_string(), + "the path reported is the one the file really landed at" + ); + // No orphan temp left behind by the fallback. + let leftovers: Vec<_> = std::fs::read_dir(archive.path()) + .unwrap() + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_string_lossy().contains("movetmp")) + .collect(); + assert!(leftovers.is_empty(), "temp file cleaned up"); + } + + #[tokio::test] + async fn a_cleanup_failure_is_retried_and_then_succeeds() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + RetryPolicy::default(), + ) + .with_source_faults( + SourceFaults::always(FsOp::RemoveFile, std::io::ErrorKind::Other).heal_after(1), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + + // Past the backoff gate, the reconciliation pass re-drives it and it completes. + worker.drive_cleanup(10_000_000, false).await.unwrap(); + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.state, ItemState::Completed); + assert!(!src.path().join("r.csv").exists()); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + assert_eq!( + fake.events_named("FileDeleted").len(), + 1, + "the success event fires exactly once, on the attempt that worked" + ); + assert!(fake.events_named("FileCleanupFailed").is_empty()); + } + + #[tokio::test] + async fn a_cleanup_failure_that_exhausts_its_budget_parks_and_reports() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let retry = RetryPolicy { + max_attempts: Some(3), + ..RetryPolicy::default() + }; + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + retry, + ) + .with_source_faults(SourceFaults::always( + FsOp::RemoveFile, + std::io::ErrorKind::Other, + )) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + worker.drive_cleanup(10_000_000, false).await.unwrap(); // attempt 2 → still retrying + assert!(fake.events_named("FileCleanupFailed").is_empty()); + worker.drive_cleanup(20_000_000, false).await.unwrap(); // attempt 3 → budget spent + + let row = store.get(INST, "r.csv").unwrap().unwrap(); + assert_eq!(row.state, ItemState::CleanupFailed); + assert_eq!(row.cleanup_attempts, 3); + assert_eq!(row.next_attempt_at, CLEANUP_NO_RETRY); + assert_eq!(row.attempts, 0, "the transfer budget was never touched"); + assert!(src.path().join("r.csv").exists()); + assert_eq!(store.stats(INST).unwrap().replicated, 0); + + let failures = fake.events_named("FileCleanupFailed"); + assert_eq!(failures.len(), 1); + assert_eq!(failures[0].body["action"], serde_json::json!("delete")); + assert_eq!(failures[0].body["attempts"], serde_json::json!(3)); + + // A parked item is no longer picked up by the ordinary reconciliation pass. + worker.drive_cleanup(i64::MAX - 1, false).await.unwrap(); + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().cleanup_attempts, + 3, + "no further automatic attempts once the budget is spent" + ); + } + + #[tokio::test] + async fn trigger_redrives_a_parked_cleanup_failure() { + // The operator path back: fix the cause, then `trigger` (which re-drives regardless of the + // gate). Modelled here by the fault table healing before the forced re-drive. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let retry = RetryPolicy { + max_attempts: Some(2), + ..RetryPolicy::default() + }; + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + retry, + ) + .with_source_faults( + SourceFaults::always(FsOp::RemoveFile, std::io::ErrorKind::Other).heal_after(2), + ); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + worker.drive_cleanup(10_000_000, false).await.unwrap(); // attempt 2 → budget spent, parked + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().next_attempt_at, + CLEANUP_NO_RETRY + ); + + worker.drive_cleanup(20_000_000, true).await.unwrap(); + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::Completed + ); + assert!(!src.path().join("r.csv").exists()); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + } + + /// Drive a file all the way to `CleanupPending` with a failing source filesystem, then hand the + /// durable state to a fresh, fault-free worker — the shape of a crash inside the write-ahead + /// cleanup window. + async fn crash_into_cleanup_pending( + store: &Arc, + src: &Path, + dst: &Path, + archive: Option<&Path>, + on_success: OnSuccess, + ) { + let mut comp = completion(on_success); + comp.archive_dir = archive.map(|p| p.to_path_buf()); + let crashing = local_worker(store.clone(), src, dst, comp, RetryPolicy::default()) + .with_source_faults( + SourceFaults::always(FsOp::Rename, std::io::ErrorKind::Other) + .and(FsOp::Copy, std::io::ErrorKind::Other) + .and(FsOp::RemoveFile, std::io::ErrorKind::Other), + ); + process_only_item(&crashing, store, 100).await; + // The prior run died between the write-ahead marker and the proof, leaving the row pending. + store + .set_state(INST, "r.csv", ItemState::CleanupPending, 100) + .unwrap(); + } + + #[tokio::test] + async fn recovery_retries_a_cleanup_pending_item_whose_source_is_still_present() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + crash_into_cleanup_pending( + &store, + src.path(), + dst.path(), + Some(archive.path()), + OnSuccess::Archive, + ) + .await; + assert!(src.path().join("r.csv").exists(), "source still on disk"); + + let (fake, events) = recording_events(); + let recovered = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + RetryPolicy::default(), + SourceFaults::default(), + ) + .with_events(events); + recovered.recover(200).await.unwrap(); + + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::Completed + ); + assert_eq!( + std::fs::read(archive.path().join("r.csv")).unwrap(), + b"evidence" + ); + assert!(!src.path().join("r.csv").exists()); + assert_eq!(fake.events_named("FileArchived").len(), 1); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + } + + #[tokio::test] + async fn recovery_completes_a_cleanup_pending_item_already_archived_before_the_crash() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + crash_into_cleanup_pending( + &store, + src.path(), + dst.path(), + Some(archive.path()), + OnSuccess::Archive, + ) + .await; + // The prior run's move DID land; the crash fell between the rename and the `Completed` write. + std::fs::rename(src.path().join("r.csv"), archive.path().join("r.csv")).unwrap(); + + let (fake, events) = recording_events(); + let recovered = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + RetryPolicy::default(), + SourceFaults::default(), + ) + .with_events(events); + recovered.recover(200).await.unwrap(); + + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::Completed, + "source absent + target present → complete" + ); + assert_eq!( + std::fs::read(archive.path().join("r.csv")).unwrap(), + b"evidence", + "the archived copy is untouched, not moved twice" + ); + assert_eq!(fake.events_named("FileArchived").len(), 1); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + } + + #[tokio::test] + async fn recovery_completes_a_cleanup_pending_item_whose_source_was_already_deleted() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + crash_into_cleanup_pending(&store, src.path(), dst.path(), None, OnSuccess::Delete).await; + // The prior run's delete DID land; the crash fell before the `Completed` write. + std::fs::remove_file(src.path().join("r.csv")).unwrap(); + + let (fake, events) = recording_events(); + let recovered = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + RetryPolicy::default(), + ) + .with_events(events); + recovered.recover(200).await.unwrap(); + + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::Completed, + "source absent after a delete intent → complete" + ); + assert_eq!(fake.events_named("FileDeleted").len(), 1); + assert_eq!(store.stats(INST).unwrap().replicated, 1); + } + + #[tokio::test] + async fn a_cleanup_failed_source_is_never_re_enqueued_by_a_rescan() { + // The source of a cleanup-failed item stays on disk, so every rescan re-discovers it. It must + // not become new work: the bytes are already delivered and verified everywhere. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let worker = local_worker( + store.clone(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + RetryPolicy::default(), + ) + .with_source_faults(SourceFaults::always( + FsOp::RemoveFile, + std::io::ErrorKind::Other, + )); + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::CleanupFailed + ); + + // A rescan re-discovers the still-present source and re-upserts it. + store.upsert_ready(INST, "r.csv", 8, 0, 300).unwrap(); + assert_eq!( + store.get(INST, "r.csv").unwrap().unwrap().state, + ItemState::CleanupFailed + ); + assert!( + store.claim_ready(INST, 10, 300).unwrap().is_empty(), + "not claimable as new work" + ); + } + + #[tokio::test] + async fn the_archive_path_reported_is_the_collision_resolved_one() { + // Under the default `suffix` policy the file lands at `r.1.csv`; the event must say so rather + // than report the computed `r.csv` that belongs to somebody else's file. + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + let archive = tempfile::tempdir().unwrap(); + std::fs::write(archive.path().join("r.csv"), b"an older run").unwrap(); + let store = store(); + enqueue_file(&store, src.path(), "r.csv", b"evidence"); + let (fake, events) = recording_events(); + let worker = archiving_worker( + store.clone(), + src.path(), + dst.path(), + Some(archive.path()), + RetryPolicy::default(), + SourceFaults::default(), + ) + .with_events(events); + + assert_eq!( + process_only_item(&worker, &store, 100).await, + ItemState::Completed + ); + let archived = fake.events_named("FileArchived"); + assert_eq!(archived.len(), 1); + assert_eq!( + archived[0].body["archivePath"].as_str().unwrap(), + archive.path().join("r.1.csv").display().to_string() + ); + assert_eq!( + std::fs::read(archive.path().join("r.1.csv")).unwrap(), + b"evidence" + ); + assert_eq!( + std::fs::read(archive.path().join("r.csv")).unwrap(), + b"an older run", + "the pre-existing file is untouched" + ); + } + + #[test] + fn verify_archived_rejects_a_target_that_does_not_match_the_source() { + let dir = tempfile::tempdir().unwrap(); + let target = dir.path().join("archived.csv"); + std::fs::write(&target, b"evidence").unwrap(); + let (_, checksum) = + hash_reader(&mut std::io::Cursor::new(b"evidence"), Algorithm::Crc32c).unwrap(); + + let ok = CleanupCtx { + on_success: OnSuccess::Archive, + src: dir.path().join("r.csv"), + archive_dir: Some(dir.path().to_path_buf()), + relpath: "r.csv".into(), + collision: Collision::Suffix, + verify: Verify::Checksum, + expected_size: 8, + expected_checksum: checksum.clone(), + }; + verify_archived(&target, &ok).expect("a matching copy verifies"); + + // Wrong size → permanent (the source is already gone; retrying cannot change the bytes). + let wrong_size = CleanupCtx { + expected_size: 99, + ..CleanupCtx { + expected_checksum: checksum.clone(), + ..clone_ctx(&ok) + } + }; + let err = verify_archived(&target, &wrong_size).unwrap_err(); + assert!(err.is_permanent(), "got {err:?}"); + + // Right size, wrong content → the checksum catches it. + std::fs::write(&target, b"tampered").unwrap(); + let err = verify_archived(&target, &ok).unwrap_err(); + assert!(err.is_permanent(), "got {err:?}"); + assert!(err.to_string().contains("mismatch")); + + // A target that is not there at all is an I/O failure, not a mismatch. + let err = verify_archived(&dir.path().join("nope.csv"), &ok).unwrap_err(); + assert!(err.to_string().contains("io:"), "got {err:?}"); + + // Under `verify: size` a content change is not inspected at all. + let size_only = CleanupCtx { + verify: Verify::Size, + ..clone_ctx(&ok) + }; + verify_archived(&target, &size_only).expect("size policy checks only the byte count"); + } + + /// `CleanupCtx` is deliberately not `Clone` in the shipping code (it is built once per attempt); + /// this rebuilds one for the struct-update syntax above. + #[cfg(test)] + fn clone_ctx(c: &CleanupCtx) -> CleanupCtx { + CleanupCtx { + on_success: c.on_success, + src: c.src.clone(), + archive_dir: c.archive_dir.clone(), + relpath: c.relpath.clone(), + collision: c.collision, + verify: c.verify, + expected_size: c.expected_size, + expected_checksum: c.expected_checksum.clone(), + } + } + + #[test] + fn decide_cleanup_uses_its_own_attempt_budget_not_the_transfer_time_budget() { + let src = tempfile::tempdir().unwrap(); + let dst = tempfile::tempdir().unwrap(); + // A transfer budget that is already fully spent — the cleanup decision must not read it. + let retry = RetryPolicy { + give_up_after_ms: Some(1), + max_attempts: None, + ..RetryPolicy::default() + }; + let worker = local_worker( + store(), + src.path(), + dst.path(), + completion(OnSuccess::Delete), + retry, + ); + assert!( + matches!( + worker.decide_cleanup(false, 0, 10_000_000_000), + RetryDecision::Retry { .. } + ), + "an expired transfer budget must not deny the first cleanup retry" + ); + // The default attempt cap still bounds it. + assert_eq!( + worker.decide_cleanup(false, DEFAULT_CLEANUP_MAX_ATTEMPTS - 1, 100), + RetryDecision::GiveUp + ); + // And a permanent error never retries. + assert_eq!(worker.decide_cleanup(true, 0, 100), RetryDecision::GiveUp); + } } diff --git a/src/state.rs b/src/state.rs index 3b92b7b..57a3549 100644 --- a/src/state.rs +++ b/src/state.rs @@ -8,8 +8,11 @@ //! inline and pushes bulk scans through `spawn_blocking`. One DB is shared across instances, keyed by //! `instance`. Every state transition is persisted *before* the side effect it authorizes (§13.2): //! `Ready → InProgress` ([`claim_ready`](StateStore::claim_ready)) precedes the transfer, `Verified` -//! ([`set_state`](StateStore::set_state)) precedes the source delete/archive, so a crash at any point -//! is recovered idempotently via [`recover_incomplete`](StateStore::recover_incomplete). +//! precedes the aggregate completion decision, and `CleanupPending` +//! ([`set_state`](StateStore::set_state)) precedes the source delete/archive — `Completed` is written +//! only once that action is proven done (DESIGN §20-I). A crash at any point is recovered +//! idempotently via [`recover_incomplete`](StateStore::recover_incomplete), which returns the +//! `InProgress`, `Verified`, and `CleanupPending` rows a prior run left behind. use std::path::Path; use std::sync::Mutex; @@ -54,6 +57,19 @@ pub trait StateStore: Send + Sync { now: i64, ) -> Result<()>; + /// Cleanup failure path (DESIGN §20-I): `cleanup_attempts += 1`, record `err`, set the cleanup + /// backoff gate in `next_attempt_at`, and move to `next_state` (`CleanupFailed`). The transfer's + /// own `attempts` are left untouched — the two retry budgets are independent. + fn record_cleanup_attempt( + &self, + instance: &str, + relpath: &str, + err: &str, + next_state: ItemState, + next_attempt_at: i64, + now: i64, + ) -> Result<()>; + /// Persist streamed-byte progress for an in-flight item. fn set_bytes_done(&self, instance: &str, relpath: &str, bytes: u64, now: i64) -> Result<()>; @@ -173,7 +189,9 @@ pub struct Stats { /// The DDL for the state DB. One DB per component data dir; instances share every table, keyed by /// `instance`. `next_attempt_at` is the P1 addition to the §14.2 schema (the backoff re-claim gate, -/// see [`WorkItem::next_attempt_at`]). `resume.blob` is a JSON-serialized [`ResumeState`] so the S3 +/// see [`WorkItem::next_attempt_at`]); `cleanup_attempts` is the source-completion retry counter +/// (DESIGN §20-I), kept separate from the transfer's `attempts` so the two budgets are independent — +/// it is also applied to an already-existing DB via [`ADDED_ITEM_COLUMNS`]. `resume.blob` is a JSON-serialized [`ResumeState`] so the S3 /// backend (P2) reuses the column for `{uploadId, completedParts}` with no migration. `dest_state` is /// the P6 addition (DESIGN §20-B): independent per-destination completion tracking for /// multi-destination fan-out, keyed the same way as `resume` — `(instance, relpath, dest)`. The DB is @@ -189,6 +207,7 @@ CREATE TABLE IF NOT EXISTS work_items( mtime_ms INTEGER NOT NULL DEFAULT 0, discovered_at INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, + cleanup_attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at INTEGER NOT NULL DEFAULT 0, last_error TEXT, bytes_done INTEGER NOT NULL DEFAULT 0, @@ -219,7 +238,12 @@ CREATE TABLE IF NOT EXISTS stats( /// The column list for every `work_items` full-row read, in [`row_to_item`] order. const ITEM_COLS: &str = "instance, relpath, state, size, discovered_at, attempts, \ - next_attempt_at, last_error, bytes_done, updated_at"; + next_attempt_at, last_error, bytes_done, updated_at, cleanup_attempts"; + +/// Columns added to `work_items` after the original release, applied to an already-existing DB with a +/// guarded `ALTER TABLE` (SQLite has no `ADD COLUMN IF NOT EXISTS`, and `CREATE TABLE IF NOT EXISTS` +/// only covers a *fresh* file — see [`SqliteStore::init`]). +const ADDED_ITEM_COLUMNS: &[&str] = &["cleanup_attempts INTEGER NOT NULL DEFAULT 0"]; /// Crash-safe durable state backed by SQLite in WAL mode (DESIGN §14). /// @@ -254,6 +278,15 @@ impl SqliteStore { conn.pragma_update(None, "foreign_keys", "ON")?; conn.busy_timeout(std::time::Duration::from_secs(5))?; conn.execute_batch(SCHEMA)?; + // `CREATE TABLE IF NOT EXISTS` covers a fresh DB and a whole new table, but not a column added + // to an existing table. Add each late column with a guarded `ALTER TABLE`: SQLite reports an + // already-present column as a plain error, which is the "nothing to do" case here. + for col in ADDED_ITEM_COLUMNS { + let name = col.split_whitespace().next().unwrap_or(col); + if !column_exists(&conn, "work_items", name)? { + conn.execute(&format!("ALTER TABLE work_items ADD COLUMN {col}"), [])?; + } + } Ok(Self { conn: Mutex::new(conn), }) @@ -267,6 +300,20 @@ impl SqliteStore { } } +/// Whether `table` already has a column named `column` (drives the guarded `ALTER TABLE` migration in +/// [`SqliteStore::init`]). +fn column_exists(conn: &Connection, table: &str, column: &str) -> Result { + let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?; + let mut rows = stmt.query([])?; + while let Some(r) = rows.next()? { + let name: String = r.get(1)?; + if name == column { + return Ok(true); + } + } + Ok(false) +} + /// Map a full `work_items` row (in [`ITEM_COLS`] order) to a [`WorkItem`]. `abs_source` is derived /// by the engine from `ingress.path` and is not a DB column, so it is reconstructed from `relpath` /// alone here (the caller re-joins the ingress root when it needs a live path). @@ -293,6 +340,7 @@ fn row_to_item(r: &Row<'_>) -> rusqlite::Result { last_error: r.get(7)?, bytes_done: r.get::<_, i64>(8)? as u64, updated_at: r.get(9)?, + cleanup_attempts: r.get::<_, i64>(10)? as u32, }) } @@ -375,6 +423,10 @@ impl StateStore for SqliteStore { WHEN work_items.state IN ('completed','quarantined','retained') AND (work_items.size <> excluded.size OR work_items.mtime_ms <> excluded.mtime_ms) THEN 0 ELSE work_items.bytes_done END, + cleanup_attempts = CASE + WHEN work_items.state IN ('completed','quarantined','retained') + AND (work_items.size <> excluded.size OR work_items.mtime_ms <> excluded.mtime_ms) + THEN 0 ELSE work_items.cleanup_attempts END, discovered_at = CASE WHEN work_items.state IN ('completed','quarantined','retained') AND (work_items.size <> excluded.size OR work_items.mtime_ms <> excluded.mtime_ms) @@ -467,6 +519,32 @@ impl StateStore for SqliteStore { Ok(()) } + fn record_cleanup_attempt( + &self, + instance: &str, + relpath: &str, + err: &str, + next_state: ItemState, + next_attempt_at: i64, + now: i64, + ) -> Result<()> { + self.lock().execute( + "UPDATE work_items + SET cleanup_attempts = cleanup_attempts + 1, last_error = ?3, state = ?4, + next_attempt_at = ?5, updated_at = ?6 + WHERE instance = ?1 AND relpath = ?2", + params![ + instance, + relpath, + err, + next_state.as_str(), + next_attempt_at, + now + ], + )?; + Ok(()) + } + fn set_bytes_done(&self, instance: &str, relpath: &str, bytes: u64, now: i64) -> Result<()> { self.lock().execute( "UPDATE work_items SET bytes_done = ?3, updated_at = ?4 @@ -494,7 +572,7 @@ impl StateStore for SqliteStore { let guard = self.lock(); let mut stmt = guard.prepare(&format!( "SELECT {ITEM_COLS} FROM work_items - WHERE instance = ?1 AND state IN ('in_progress', 'verified') + WHERE instance = ?1 AND state IN ('in_progress', 'verified', 'cleanup_pending') ORDER BY discovered_at ASC" ))?; let rows = stmt.query_map(params![instance], row_to_item)?; @@ -921,20 +999,150 @@ mod tests { } #[test] - fn recover_incomplete_returns_in_progress_and_verified_only() { + fn recover_incomplete_returns_the_non_terminal_in_flight_states_only() { let (s, _d) = temp_store(); - for (rel, disc) in [("a", 1), ("b", 2), ("c", 3), ("d", 4)] { + for (rel, disc) in [("a", 1), ("b", 2), ("c", 3), ("d", 4), ("e", 5), ("f", 6)] { s.upsert_ready(INST, rel, 1, 0, disc).unwrap(); } s.set_state(INST, "a", ItemState::InProgress, 10).unwrap(); s.set_state(INST, "b", ItemState::Verified, 10).unwrap(); s.set_state(INST, "c", ItemState::Completed, 10).unwrap(); + // A crash inside the write-ahead cleanup window must be recovered too (DESIGN §20-I). + s.set_state(INST, "e", ItemState::CleanupPending, 10) + .unwrap(); + // A cleanup FAILURE carries its own backoff gate and is re-driven by the cleanup pass on the + // next tick, not by crash recovery — so it must NOT appear here. + s.set_state(INST, "f", ItemState::CleanupFailed, 10) + .unwrap(); // "d" stays Ready. let rec = s.recover_incomplete(INST).unwrap(); assert_eq!( rec.iter().map(|i| i.relpath.as_str()).collect::>(), - vec!["a", "b"] + vec!["a", "b", "e"] + ); + } + + #[test] + fn record_cleanup_attempt_counts_separately_from_transfer_attempts() { + // DESIGN §20-I: the two retry budgets are independent, so a cleanup attempt must never disturb + // the transfer's `attempts` (which `get-status` reports and the transfer backoff reads). + let (s, _d) = temp_store(); + s.upsert_ready(INST, "f", 10, 0, 1).unwrap(); + s.record_attempt(INST, "f", "transfer boom", ItemState::Failed, 100, 5) + .unwrap(); + s.record_attempt(INST, "f", "transfer boom", ItemState::Failed, 200, 6) + .unwrap(); + + s.record_cleanup_attempt(INST, "f", "archive boom", ItemState::CleanupFailed, 900, 10) + .unwrap(); + let it = s.get(INST, "f").unwrap().unwrap(); + assert_eq!(it.state, ItemState::CleanupFailed); + assert_eq!(it.attempts, 2, "transfer attempts untouched"); + assert_eq!(it.cleanup_attempts, 1); + assert_eq!( + it.next_attempt_at, 900, + "the cleanup gate reuses next_attempt_at" ); + assert_eq!(it.last_error.as_deref(), Some("archive boom")); + assert_eq!(it.updated_at, 10); + + s.record_cleanup_attempt( + INST, + "f", + "archive boom 2", + ItemState::CleanupFailed, + 950, + 11, + ) + .unwrap(); + let it = s.get(INST, "f").unwrap().unwrap(); + assert_eq!(it.cleanup_attempts, 2); + assert_eq!(it.attempts, 2); + } + + #[test] + fn rediscovery_never_reenqueues_a_cleanup_state() { + // DESIGN §20-I / FR-REL-1: the source of a `CleanupPending`/`CleanupFailed` item is still on + // disk, so every rescan re-discovers it. Neither state is terminal, so `upsert_ready` must + // preserve the row rather than resurrect an already-replicated file as new work. + let (s, _d) = temp_store(); + for (rel, state) in [ + ("pending", ItemState::CleanupPending), + ("failed", ItemState::CleanupFailed), + ] { + s.upsert_ready(INST, rel, 10, 111, 1).unwrap(); + s.set_state(INST, rel, state, 2).unwrap(); + // Re-discovered unchanged, and re-discovered with a different signature: neither resets. + s.upsert_ready(INST, rel, 10, 111, 3).unwrap(); + assert_eq!(s.get(INST, rel).unwrap().unwrap().state, state); + s.upsert_ready(INST, rel, 999, 222, 4).unwrap(); + assert_eq!( + s.get(INST, rel).unwrap().unwrap().state, + state, + "{rel}: a cleanup-state row is never re-enqueued as new work" + ); + } + assert!( + s.claim_ready(INST, 10, 100).unwrap().is_empty(), + "and neither is claimable" + ); + } + + #[test] + fn cleanup_attempts_reset_when_a_changed_file_reuses_a_completed_relpath() { + // The rotating-filename producer pattern: a genuinely new file at a completed relpath starts + // with a fresh cleanup budget, not the prior file's spent one. + let (s, _d) = temp_store(); + s.upsert_ready(INST, "report.csv", 10, 111, 1).unwrap(); + s.record_cleanup_attempt(INST, "report.csv", "boom", ItemState::CleanupFailed, 500, 2) + .unwrap(); + s.set_state(INST, "report.csv", ItemState::Completed, 3) + .unwrap(); + s.upsert_ready(INST, "report.csv", 10, 222, 10).unwrap(); + let it = s.get(INST, "report.csv").unwrap().unwrap(); + assert_eq!(it.state, ItemState::Ready); + assert_eq!( + it.cleanup_attempts, 0, + "fresh cleanup budget for a new file" + ); + } + + #[test] + fn a_db_written_before_the_cleanup_column_is_migrated_on_open() { + // `CREATE TABLE IF NOT EXISTS` cannot add a column to an existing table, so opening a DB from + // an earlier build must apply the guarded ALTER (see `ADDED_ITEM_COLUMNS`) rather than fail + // every subsequent `SELECT`. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("legacy.db"); + { + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TABLE work_items( + instance TEXT NOT NULL, + relpath TEXT NOT NULL, + state TEXT NOT NULL, + size INTEGER NOT NULL, + mtime_ms INTEGER NOT NULL DEFAULT 0, + discovered_at INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + bytes_done INTEGER NOT NULL DEFAULT 0, + updated_at INTEGER NOT NULL, + PRIMARY KEY(instance, relpath)); + INSERT INTO work_items(instance, relpath, state, size, discovered_at, updated_at) + VALUES ('inst-a', 'old.csv', 'ready', 7, 1, 1);", + ) + .unwrap(); + } + let s = SqliteStore::open(&path).unwrap(); + let it = s.get(INST, "old.csv").unwrap().unwrap(); + assert_eq!(it.size, 7); + assert_eq!(it.cleanup_attempts, 0, "back-filled at the column default"); + // The migration is idempotent across reopens. + drop(s); + let s = SqliteStore::open(&path).unwrap(); + assert_eq!(s.claim_ready(INST, 10, 100).unwrap().len(), 1); } #[test] diff --git a/tests/azure_azurite.rs b/tests/azure_azurite.rs index 2b3eab5..841890a 100644 --- a/tests/azure_azurite.rs +++ b/tests/azure_azurite.rs @@ -201,6 +201,7 @@ fn work_item(src_root: &Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/tests/gcs_fakegcs.rs b/tests/gcs_fakegcs.rs index a7797bc..1305745 100644 --- a/tests/gcs_fakegcs.rs +++ b/tests/gcs_fakegcs.rs @@ -221,6 +221,7 @@ fn work_item(src_root: &Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/tests/http_inprocess.rs b/tests/http_inprocess.rs index 87e89e2..7dbf47f 100644 --- a/tests/http_inprocess.rs +++ b/tests/http_inprocess.rs @@ -360,6 +360,7 @@ fn work_item(src_root: &std::path::Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/tests/p1_engine.rs b/tests/p1_engine.rs index 4df1a8a..d415f76 100644 --- a/tests/p1_engine.rs +++ b/tests/p1_engine.rs @@ -246,6 +246,18 @@ impl StateStore for FailCompletedOnceStore { self.inner .record_attempt(instance, relpath, err, next_state, next_attempt_at, now) } + fn record_cleanup_attempt( + &self, + instance: &str, + relpath: &str, + err: &str, + next_state: ItemState, + next_attempt_at: i64, + now: i64, + ) -> ReplResult<()> { + self.inner + .record_cleanup_attempt(instance, relpath, err, next_state, next_attempt_at, now) + } fn set_bytes_done(&self, instance: &str, relpath: &str, bytes: u64, now: i64) -> ReplResult<()> { self.inner.set_bytes_done(instance, relpath, bytes, now) } @@ -787,18 +799,21 @@ async fn crash_after_source_delete_before_completed_recovers_exactly_once() { )); // Phase A — one tick: deliver + verify succeed, the source is deleted, but persisting Completed - // fails (the injected crash). The item is left Verified: source gone, dest written, not counted. + // fails (the injected crash). The item is left CleanupPending: source gone, dest written, not + // counted. `CleanupPending` is the write-ahead marker written immediately before the source side + // effect (DESIGN §20-I), so it is exactly what a crash in this window leaves behind. inst.tick(1_000).await; assert_eq!(std::fs::read(dst.path().join("once.dat")).unwrap(), PAYLOAD, "delivered"); assert!(!src.path().join("once.dat").exists(), "source side effect ran before Completed persist"); assert_eq!( store.get("once", "once.dat").unwrap().unwrap().state, - ItemState::Verified, - "Verified persisted before the source side effect; Completed not yet written" + ItemState::CleanupPending, + "the cleanup intent is persisted before the source side effect; Completed not yet written" ); assert_eq!(store.stats("once").unwrap().replicated, 0, "not counted before Completed"); - // Phase B — restart: recovery re-verifies the destination (still present) and completes the item. + // Phase B — restart: recovery re-evaluates the cleanup against observed state (the source is + // already gone, so the delete landed) and completes the item. let cancel = CancellationToken::new(); let handle = { let inst = inst.clone(); diff --git a/tests/s3_floci.rs b/tests/s3_floci.rs index 2dedf97..a36ab25 100644 --- a/tests/s3_floci.rs +++ b/tests/s3_floci.rs @@ -215,6 +215,7 @@ fn work_item(src_root: &Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/tests/s3_real.rs b/tests/s3_real.rs index e30cc98..f64f73e 100644 --- a/tests/s3_real.rs +++ b/tests/s3_real.rs @@ -83,6 +83,7 @@ fn work_item(root: &Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0, diff --git a/tests/sftp_atmoz.rs b/tests/sftp_atmoz.rs index ffbf34b..563c745 100644 --- a/tests/sftp_atmoz.rs +++ b/tests/sftp_atmoz.rs @@ -198,6 +198,7 @@ fn work_item(src_root: &Path, rel: &str) -> WorkItem { size, discovered_at: 0, attempts: 0, + cleanup_attempts: 0, next_attempt_at: 0, last_error: None, bytes_done: 0,