diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 7b70a7bd..5bc93c9b 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -1683,12 +1683,11 @@ lane; demand-gated on a first enterprise Windows/AD deployment. ## 114. Directory validation toggle (perform vs suppress startup validation) -> πŸ”’ **Re-scored 2026-08-03 β†’ DEMAND-GATE.** Value **6/10** Β· Difficulty **3/10** Β· _quick win_. The old score's "clean workaround via the on-demand test probe" does not exist in the direction that remains β€” both destinations' `test_connection` *create* the target directory, so the probe cannot answer the question the toggle asks, which is what lifts this off the parity-with-a-workaround band; the silent-ignore half is closed (PR #162 raises `WiringError` on `File(validate_directory=True)` for an outbound), leaving only the validation hook β€” a `validate_startup` on the `DestinationConnector` contract plus a runner outbound start-path call, mirroring the source seam already at `transports/base.py:436`. _(was 5/10 Β· 2/10.)_ -> **On-trigger / demand-gate.** Numbered for tracking only β€” build when the trigger below fires (β€œdemand-gate, don’t schedule”). +> βœ… **SHIPPED 2026-08-11 β€” the outbound half is built; #114 is complete.** `DestinationConnector.validate_startup()` is a **default no-op**, so the other eleven destination connectors are untouched and this is not a protocol change that ripples. `FileDestination` and `RemoteFileDestination` override it, and the runner awaits it in `_start_outbound` and `_ensure_destination_built` **inside the existing [ADR 0031](adr/0031-startup-connection-fault-isolation.md) isolation `try`** β€” so a refusal is a `failed` lane with no connector, whose delivery worker **still spawns**: routed rows are retained, retried and buildup-alerted rather than dropped, and the count-and-log invariant holds. **The defect, stated in the conditional:** `FileDestination` previously `mkdir`-ed its target on write, so a deploying operator who typo'd `remote_dir` would get a directory silently **created** and messages delivered into it β€” the feed reading healthy while writing to the wrong place, with nothing reporting it. `validate_directory` becomes a **both-directions** option and the 2026-08-03 outbound `WiringError` is **removed** β€” it existed only because no destination read the setting. Under the toggle nothing is ever created: not at start, not on write, and not by `POST /connections/{name}/test`. **One deliberate change reaches every existing outbound, toggled or not:** a target directory the engine actually had to **create** now logs a `WARNING`, so a delivery into an invented path is no longer indistinguishable from a normal one. ADR 0031 is amended, with its 2026-08-03 follow-on marked superseded. **This item's own 6/10 rationale was re-measured and is FALSE in this direction.** It claimed a *"clean workaround via the on-demand test probe"* β€” but both destinations' `test_connection` **create the directory**, so the act of asking changes the answer. An item's scoring rationale falsified by execution, the same class as #1011's refuted premise. -> **AMENDED 2026-08-03 β€” the INBOUND half and the outbound WIRING REJECTION are BUILT; only the outbound validation HOOK remains.** Adversarial verification refuted a full close. **BUILT 2026-07-28:** `validate_directory` on the File/RemoteFile source (`messagefoundry/transports/file.py:311`, `remotefile.py:735`) with its opt-in at-start check (`file.py:389-398`) β€” a no-mkdir probe that reports the connection `failed` at start rather than deferring to first poll (`file.py:170`). **BUILT 2026-08-03:** the option on an **outbound** is now a **`WiringError` at bind** (`build_outbound_connection`, `messagefoundry/config/wiring.py`) instead of being accepted and silently ignored. That is the single choke point both code-first `outbound()` and the `connections.toml` loader (ADR 0007) pass through, so one guard covers both authoring surfaces; it is truthy-only, so the `False` the factories always write into settings is unaffected and every outbound authored today builds byte-identically. +> **AMENDED 2026-08-03 β€” SUPERSEDED 2026-08-11 by the banner above, which closed the remaining outbound half. Kept as the record of how the item was built in stages.** At the time it read *"only the outbound validation HOOK remains"*, which was true then and is not now. **One statement below was also REVERSED by the close:** the 2026-08-03 outbound `WiringError` described as BUILT was subsequently **removed**, because it existed only to reject a setting no destination read -- and a destination now reads it. Adversarial verification refuted a full close. **BUILT 2026-07-28:** `validate_directory` on the File/RemoteFile source (`messagefoundry/transports/file.py:311`, `remotefile.py:735`) with its opt-in at-start check (`file.py:389-398`) β€” a no-mkdir probe that reports the connection `failed` at start rather than deferring to first poll (`file.py:170`). **BUILT 2026-08-03:** the option on an **outbound** is now a **`WiringError` at bind** (`build_outbound_connection`, `messagefoundry/config/wiring.py`) instead of being accepted and silently ignored. That is the single choke point both code-first `outbound()` and the `connections.toml` loader (ADR 0007) pass through, so one guard covers both authoring surfaces; it is truthy-only, so the `False` the factories always write into settings is unaffected and every outbound authored today builds byte-identically. > -> ⚠️ **REMAINDER: the outbound validation HOOK β€” and this item's scoring rationale is WRONG for that direction.** `DestinationConnector` still has no `validate_startup` hook and `FileDestination` still `mkdir`s on write. The "clean workaround via the on-demand test probe" cited in the score above **does not exist on an outbound**: both destinations' `test_connection` *create* the target directory (see ADR 0031's 2026-08-03 follow-on for the call chain), so nothing shipped can tell "the directory exists" from "I just made it" β€” a typo'd target path is fabricated and every message reports delivered. **Re-score against that.** And if the hook is built, build it **together with** suppressing the mkdir-on-write under the flag: a start-time-only check leaves the run-time fabrication intact under a setting name that promises otherwise. +> ⚠️ **WHAT THE REMAINDER WAS, and the scoring rationale it refuted β€” BUILT 2026-08-11, so read the two claims below in the PAST TENSE.** `DestinationConnector` **had** no `validate_startup` hook and `FileDestination` **mkdir'd** on write. The "clean workaround via the on-demand test probe" cited in the score above **does not exist on an outbound**: both destinations' `test_connection` *create* the target directory (see ADR 0031's 2026-08-03 follow-on for the call chain), so nothing shipped can tell "the directory exists" from "I just made it" β€” a typo'd target path is fabricated and every message reports delivered. **That instruction was a build gate, and it was HONOURED:** it required that the hook be built **together with** suppressing the mkdir-on-write, because a start-time-only check would leave the run-time fabrication intact under a setting name that promises otherwise. Under the shipped toggle nothing is created at start, on write, or by `POST /connections/{name}/test`. **Cluster:** Connections & Transports. **Priority:** P3. **Verdict:** demand-gate. **Severity (vs Corepoint):** minor. diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index 8a989631..2995d88f 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -648,7 +648,7 @@ def route(msg): | `sort` | in | `name` | process order: `name` or `mtime` | | `recursive` | in | `false` | also scan subdirectories | | `max_file_bytes` | in | `16 MiB` | route files larger than this to the error dir instead of reading them into memory (OOM guard). `None`/`0` = unlimited. | -| `validate_directory` | in | `false` | validate the poll directory **at startup** (#114): a missing/unusable dir reports the connection **`failed`** (ADR 0031) instead of the default deferral to run time. No mkdir β€” a merely-missing dir fails. A `leave` source validates read-only (a read-only share passes); `move`/`delete` also require write. **Inbound only** β€” on an outbound it is a `WiringError` at bind (an outbound target directory is never validated at startup; it is `mkdir`ed on write). | +| `validate_directory` | both | `false` | validate the directory **at startup** (#114): a missing/unusable dir reports the connection **`failed`** (ADR 0031) instead of the default deferral to run time. **No mkdir** β€” a merely-missing dir fails. **In:** a `leave` source validates read-only (a read-only share passes); `move`/`delete` also require write. **Out:** the target must already exist and accept a write, and is then **never created** β€” not at start, not on write (a delivery into a vanished dir fails retryably instead), and not by `POST /connections/{name}/test`. Left off (the default) the outbound target is still created on first write, but the creation is now logged as a `WARNING`. | | `processed_subdir` / `error_subdir` | in | `.processed` / `.error` | where read/failed files go | | `filename` | out | `{MSH-10}.hl7` | output name (supports `{HL7-path}` placeholders). Resolved values are sanitized to a **single safe filename** β€” path separators/unsafe chars stripped, leading dots removed, and `.`/`..`/reserved device names fall back β€” so a message field can never write outside the directory. | | `overwrite` | out | `false` | overwrite vs. uniquify a name collision (collisions are resolved by an **atomic** exclusive create, so concurrent writes never clobber) | @@ -871,7 +871,7 @@ poll/write shape against a remote server, selected by an internal `protocol` set | `min_age_seconds` | in | `0.0` | **accepted but not honoured on a remote source today** β€” the connector never reads it (a remote directory listing carries no reliable mtime). Only `File(...)` implements it; use `after_read`/the partner's own write-then-rename to avoid partial reads. | | `after_read` | in | `move` | `move` (β†’ `processed_subdir`), `delete`, or `leave` (process **in place**, #142 β€” a durable dedup ledger keyed on a hash of the **full remote path** + size ensures a left file is ingested once) | | `max_file_bytes` | in | `16 MiB` | move a file larger than this to `error_subdir` instead of retrieving it (OOM guard). `None`/`0` = unlimited. | -| `validate_directory` | in | `false` | validate `remote_dir` **at startup** (#114): unreachable/unusable reports the connection **`failed`** (ADR 0031) instead of deferring to run time. **Inbound only** β€” on an outbound it is a `WiringError` at bind (the upload dir is `ensure_dir`ed on write, never validated at startup). | +| `validate_directory` | both | `false` | validate `remote_dir` **at startup** (#114): unreachable/unusable reports the connection **`failed`** (ADR 0031) instead of deferring to run time. The probe is a **listing** β€” it never creates. **Out:** the upload dir is then never `ensure_dir`ed either, on send or by `POST /connections/{name}/test`; an upload into a vanished dir fails **retryably** rather than dead-lettering on the partner's permanent no-such-dir. Left off (the default) the upload dir is still created on first send, but the creation is now logged as a `WARNING`. | | `processed_subdir` / `error_subdir` | in | `.processed` / `.error` | where read / failed files go | | `filename` | out | `{MSH-10}.hl7` | upload name (supports `{HL7-path}` placeholders, sanitized to a **single safe filename** exactly as `File(...)`) | | `overwrite` | out | `false` | overwrite vs. uniquify a name collision (never a silent clobber) | diff --git a/docs/adr/0031-startup-connection-fault-isolation.md b/docs/adr/0031-startup-connection-fault-isolation.md index b3550d40..be6580e4 100644 --- a/docs/adr/0031-startup-connection-fault-isolation.md +++ b/docs/adr/0031-startup-connection-fault-isolation.md @@ -198,13 +198,77 @@ The equivalent outbound (FileDestination) is out of scope here β€” it already `m the on-demand `POST /connections/{name}/test` probe. **Follow-on (2026-08-03, BACKLOG #114) β€” the outbound rejects the option rather than ignoring it.** -Because `File()`/`Sftp()`/`Ftp()` are single factories serving both directions, the option above could -be *written* onto an outbound, where nothing reads it β€” accepted and silently ignored. That is now a -**`WiringError` at bind** in `build_outbound_connection`, the one choke point both the code-first -`outbound()` and the `connections.toml` loader (ADR 0007) pass through. Truthy-only, so the `False` -the factories always write is unaffected and every outbound authored today builds byte-identically. -The outbound *validation hook* itself (`DestinationConnector.validate_startup`) remains **out of -scope** and deferred β€” note that the "on-demand test probe" workaround cited above is **inbound-only -in effect**: `FileDestination.test_connection` β†’ `_probe_dir_writable` and -`RemoteFileDestination.test_connection` β†’ `ensure_dir` both **create** the target directory, so on an -outbound no shipped mechanism can distinguish "the directory exists" from "I just made it." +**Superseded by the 2026-08-10 amendment below, which builds the hook and removes this `WiringError`; +kept for the reasoning, which still holds.** Because `File()`/`Sftp()`/`Ftp()` are single factories +serving both directions, the option above could be *written* onto an outbound, where nothing read it β€” +accepted and silently ignored. That was made a **`WiringError` at bind** in +`build_outbound_connection`, the one choke point both the code-first `outbound()` and the +`connections.toml` loader (ADR 0007) pass through. The outbound *validation hook* itself +(`DestinationConnector.validate_startup`) was out of scope here β€” note that the "on-demand test probe" +workaround cited above is **inbound-only in effect**: `FileDestination.test_connection` β†’ +`_probe_dir_writable` and `RemoteFileDestination.test_connection` β†’ `ensure_dir` both **create** the +target directory, so on an outbound no shipped mechanism could distinguish "the directory exists" from +"I just made it." + +## Amendment (2026-08-10, BACKLOG #114) β€” the outbound half: `DestinationConnector.validate_startup` + +**Status:** Accepted (owner go β€” build the remainder). Built in the same change. + +**Context.** The 2026-07-17 amendment deferred the outbound hook on two grounds: the destination +"already `mkdir`s on write", and it "has the on-demand `POST /connections/{name}/test` probe". The +2026-08-03 follow-on already withdrew the second (both destinations' `test_connection` **create** the +directory β€” re-measured against the shipped code before this amendment was written, on a missing +directory, for both FILE and REMOTEFILE). This withdraws the first: `mkdir`-on-write is not a weaker +form of validation, it is the defect. A typo'd `directory`/`remote_dir` does not fail β€” it is +**created**, and every message delivered into it is counted and logged as delivered, because it was. +On a first deployment that is a feed landing in a path nobody is watching with no error anywhere. +(Nothing is misdelivering today; there are zero deployments β€” see CLAUDE.md Β§0.) + +**Decision.** Three parts. + +1. **`DestinationConnector.validate_startup()`**, defaulting to a **no-op** β€” the exact shape of the + `SourceConnector` hook above, so the other eleven destination connectors are untouched and this is + not a protocol change that ripples. `FileDestination` and `RemoteFileDestination` override it. The + runner awaits it in `_start_outbound` immediately after the connector is built, **inside the + existing ADR-0031 isolation `try`**: a `DestinationStartupError` therefore takes the same path as a + build failure β€” the lane is recorded `failed` with **no live connector**, its delivery worker is + **still spawned**, and rows routed to it are retried + buildup-alerted, never dropped. On an + outbound, "invalid means not-started" *is* that degraded-lane state, so Β§1's reliability and + count-and-log invariants are preserved rather than re-argued. The same call is made on the operator + start path (`_ensure_destination_built`, which already isolates rather than raises). It is **not** + made on the reload path, whose stated invariant is that a connector build there cannot fail (intake + is quiesced at that point, so a raise would strand the swap). + +2. **`validate_directory` becomes a both-directions option** on `File`/`Sftp`/`Ftp`, and the + 2026-08-03 outbound `WiringError` is **removed**. That guard existed for exactly one reason β€” no + destination read the setting β€” and that reason is now gone; keeping it would mean shipping the hook + behind a second, differently-named knob. Default stays `false`, so every outbound authored today + builds and runs byte-identically. + +3. **A created directory is loud.** Under the default (defer) arm the target is still created on + write and the delivery still succeeds β€” but a create that actually happened now logs a `WARNING` + naming the path. This is the half that applies to every existing outbound, because it is the + default arm: the failure mode being closed is silence, not the creation itself. + +Three deliberate details, each the mirror of a source-side one: + +- **No-create at every asking point.** `validate_startup` uses `_probe_dir_startup` (FILE) or a + `list_dir` (REMOTEFILE) β€” never `_probe_dir_writable`/`ensure_dir`, both of which create. Under + `validate_directory=true` `test_connection` switches to the same no-create probes, because + otherwise the operator's own `POST /connections/{name}/test` would silently repair the typo the + toggle exists to catch and the next restart would then validate clean. +- **No-create at delivery time too, under the toggle.** "This directory must exist" has to keep + meaning that after start, so `_write`/`_upload` do not create it either: a share that vanished + mid-run fails the send **retryably** and the lane backs off and self-heals. The REMOTEFILE arm + pre-checks with a `list_dir` specifically to reclassify β€” an SFTP/FTP no-such-dir is a **permanent** + error, so letting the upload fail naturally would dead-letter live traffic over a merely-unmounted + share. It costs one extra round trip per delivery, on the opt-in path only. +- **The default arm still serves the item's own trigger.** An intermittently-available directory must + **not** fail startup β€” which is why the toggle is opt-in and defaults to defer in both directions. + +**Consequences.** Additive; a graph that never sets `validate_directory` on an outbound behaves as +before apart from the create-on-write WARNING. The FILE default path's syscall count is unchanged +(`mkdir(parents=True, exist_ok=True)` already probed `is_dir()` on its `FileExistsError` branch, which +is the common one). `_RemoteClient.ensure_dir` now reports whether it created β€” a module-private +contract with two implementations. No new schema and no new dependency; a refusal rides the existing +`_failed`/`failed` surfacing and the `connection_stopped` alert. diff --git a/docs/testing/master-test-plan/02-pipeline-reliability.md b/docs/testing/master-test-plan/02-pipeline-reliability.md index 3b356b3b..bbc5f0e9 100644 --- a/docs/testing/master-test-plan/02-pipeline-reliability.md +++ b/docs/testing/master-test-plan/02-pipeline-reliability.md @@ -83,7 +83,7 @@ chapter's own row**. This chapter's area = `{FCP:PIPE-1..19}` βˆͺ | `tests/test_reingress.py` (22) | ADR 0013 Inc 2: `Stage.RESPONSE` lane keying, reset recovery, the finalizer seeing the pending response row, loopback wiring guards, exactly-once `ingress_handoff`, depth-cap dead-letter, corrupt-ref no-loop, end-to-end response worker, `response_get` after retention purge | | `tests/test_passthrough.py` (25) + `tests/test_passthrough_graph.py` | ADR 0038 internal PT connector + child-message re-ingress in the same `transform_handoff` transaction | | `tests/test_not_deployed.py` (26) + 5 cases each in `test_sqlserver_store.py` / `test_postgres_store.py` | ADR 0111: declined Sends recorded as `not_deployed` events in the **same** handoff txn; the finalizer emitting `NOT_DEPLOYED` rather than `FILTERED` for an all-declined message | -| `tests/test_startup_fault_isolation.py` (7) | ADR 0031: duplicate-port loser isolated, reserved-API-port inbound isolated, failed outbound isolated β†’ retries β†’ recovers, File `validate_directory` isolate vs defer, valid graph starts clean, `/connections` reports degraded | +| `tests/test_startup_fault_isolation.py` (9) | ADR 0031: duplicate-port loser isolated, reserved-API-port inbound isolated, failed outbound isolated β†’ retries β†’ recovers, File `validate_directory` isolate vs defer on the inbound **and on the outbound** (#114 β€” a refused outbound target is `failed` with its routed rows retained pending and the directory never fabricated; the deferred one comes up clean and logs the directory it had to create), valid graph starts clean, `/connections` reports degraded | | `tests/test_inline_fast_path.py` (11; also on real SS + PG via `ci.yml:690` / `:820`) | ADR 0057 gates: off β‡’ split path; multi-handler / filtering / state-op / lookup-graph all fall back; internal-error policy; crash-after-claim pure re-run; post-commit idempotent no-op; the G6 finite-attempts ceiling on the inline path | | `tests/test_crit2_inline_doc_drift.py` (4) | `InboundConnection.inline` and the `inbound()` factory both default `False`; the live `_router_worker` really does call `store.handoff(` (doc-drift tripwire) | | `tests/test_replay_purity.py` (7) | `RouteOutcome` value-equality replay harness: a pure Handler is byte-identical; a pinned `current_ingest_time` is pure; module-global / wall-clock / `uuid4` / impure-router all **diverge** β€” the harness has teeth | diff --git a/messagefoundry/config/wiring.py b/messagefoundry/config/wiring.py index 56f61eb5..5625e718 100644 --- a/messagefoundry/config/wiring.py +++ b/messagefoundry/config/wiring.py @@ -1675,7 +1675,7 @@ def File( sort: str = "name", # inbound: process order β€” "name" | "mtime" recursive: bool = False, # inbound: also scan subdirectories max_file_bytes: int | None = 16 * 1024 * 1024, # inbound: skip files over this (OOM guard) - validate_directory: bool = False, # inbound ONLY (a WiringError on an outbound, #114): fail-fast at start on a missing/unusable dir; default defers to run time + validate_directory: bool = False, # both directions (#114): fail-fast at start on a missing/unusable dir, and never create it; default defers to run time overwrite: bool = False, # outbound: overwrite vs. uniquify a name collision processed_subdir: str = ".processed", error_subdir: str = ".error", @@ -1699,10 +1699,13 @@ def File( ``after_read`` (inbound) chooses the source-file disposition: ``move`` (β†’ ``processed_subdir``, the default), ``delete``, or ``leave`` β€” **process in place** for a read-only share / a directory another system owns (#142; a HASHED per-file ledger dedups so a left file is ingested once). - ``validate_directory`` (**inbound only**, #114) makes a missing/unusable directory **fail startup** - (the connection is reported ``failed``) instead of the default deferral to run time; a ``leave`` - source validates read-only (a read-only share passes). On an **outbound** it raises a - :class:`WiringError` at bind β€” no destination reads it, so accepting it would be a silent no-op. + ``validate_directory`` (#114, **both directions**) makes a missing/unusable directory **fail + startup** (the connection is reported ``failed``) instead of the default deferral to run time. On an + **inbound** a ``leave`` source validates read-only (a read-only share passes) while ``move``/ + ``delete`` also need write. On an **outbound** the target must already exist and accept a write, and + the directory is then **never created** β€” not at start, not on write, and not by the on-demand + ``POST /connections/{name}/test`` probe. Leaving it off keeps the default: the outbound target is + created on first write (now logged as a WARNING when it actually had to be created). ``credential_username`` / ``credential_domain`` / ``credential_password`` (ADR 0132, #111) give the endpoint an **alternate Windows identity** for a UNC/SMB share, distinct from the engine service @@ -2654,7 +2657,7 @@ def Sftp( ] = "move", # inbound: "move" (to processed_subdir) | "delete" | "leave" (process in place, #142) min_age_seconds: float = 0.0, # inbound: skip files modified within this window (partial writes) max_file_bytes: int | None = 16 * 1024 * 1024, # inbound: skip files over this (OOM guard) - validate_directory: bool = False, # inbound ONLY (a WiringError on an outbound, #114): fail-fast at start on an unreachable remote dir + validate_directory: bool = False, # both directions (#114): fail-fast at start on an unreachable remote dir, and never create it overwrite: bool = False, # outbound: overwrite vs. uniquify a name collision processed_subdir: str = ".processed", error_subdir: str = ".error", @@ -2669,7 +2672,12 @@ def Sftp( refused) β€” accepting an unknown key needs ``MEFOR_ALLOW_INSECURE_TLS``. Put secrets (``password``/ ``private_key``/``key_password``) in ``env()``. The host is gated by ``[egress].allowed_remote`` (both directions). At-least-once: an upload may re-send and a poll may re-emit, so downstreams - **must be idempotent**.""" + **must be idempotent**. + + ``validate_directory`` (#114, both directions) makes an unreachable/missing ``remote_dir`` **fail + startup** β€” the connection is reported ``failed`` β€” instead of the default deferral to run time; on + an outbound it additionally stops the upload directory from ever being created (on send, or by the + on-demand test probe). Off by default: an intermittently-available remote dir must still start.""" return ConnectionSpec( ConnectorType.REMOTEFILE, { @@ -2714,7 +2722,7 @@ def Ftp( ] = "move", # inbound: "move" (to processed_subdir) | "delete" | "leave" (process in place, #142) min_age_seconds: float = 0.0, # inbound: skip files modified within this window (partial writes) max_file_bytes: int | None = 16 * 1024 * 1024, # inbound: skip files over this (OOM guard) - validate_directory: bool = False, # inbound ONLY (a WiringError on an outbound, #114): fail-fast at start on an unreachable remote dir + validate_directory: bool = False, # both directions (#114): fail-fast at start on an unreachable remote dir, and never create it overwrite: bool = False, # outbound: overwrite vs. uniquify a name collision processed_subdir: str = ".processed", error_subdir: str = ".error", @@ -2727,7 +2735,8 @@ def Ftp( plain ``ftp`` is **refused** unless ``MEFOR_ALLOW_INSECURE_TLS`` is set (use ``tls=True`` for FTPS, or :func:`Sftp`). FTPS encrypts the control + data channels, so credentials are fine there. Put secrets (``password``) in ``env()``. The host is gated by ``[egress].allowed_remote`` (both - directions). At-least-once β†’ downstreams **must be idempotent**.""" + directions). At-least-once β†’ downstreams **must be idempotent**. ``validate_directory`` behaves + exactly as it does on :func:`Sftp`.""" return ConnectionSpec( ConnectorType.REMOTEFILE, { @@ -4154,23 +4163,11 @@ def build_outbound_connection( f"outbound connection {name!r}: {kind} outbound requires a host (the downstream peer), " f"e.g. {kind.title()}(host=..., port=...)." ) - if spec.type in (ConnectorType.FILE, ConnectorType.REMOTEFILE) and spec.settings.get( - "validate_directory" - ): - # BACKLOG #114: validate_directory is INBOUND-ONLY. Only SourceConnector carries the - # validate_startup hook the runner awaits at bind (pipeline/wiring_runner.py) β€” no destination - # reads the setting, and DestinationConnector has no equivalent hook. Set on an outbound it was - # accepted and silently ignored, so an operator asking for fail-fast validation got none and saw - # no error. It is caught HERE rather than in File()/Sftp()/Ftp() because those factories serve - # BOTH directions and cannot know which one they are β€” only the bind does. (Same choke-point - # reasoning as the capture_response guard below, which spells it out.) Truthy-only, so the - # `False` the factories always write into settings keeps building byte-identically. - kind = spec.type.value.upper() - raise WiringError( - f"outbound connection {name!r}: validate_directory is an inbound-only option β€” an outbound " - f"{kind} never reads it, so requesting it here would be a silent no-op. Remove it. " - "(Startup validation of an outbound target directory is not implemented; see BACKLOG #114.)" - ) + # BACKLOG #114: validate_directory was rejected here while it was INBOUND-ONLY (no destination read + # it, and DestinationConnector had no validate_startup hook, so accepting it was a silent no-op). + # Both halves are built now β€” DestinationConnector.validate_startup, overridden by the FILE and + # REMOTEFILE destinations and awaited on the runner's outbound start path β€” so the option is + # honoured in both directions and there is nothing left to refuse. _check_metadata(name, metadata) # ADR 0013 Increment 2: reingress_to (route this outbound's reply back as a new inbound message) # IMPLIES capture (the reply must be captured to re-ingress it). Force capture_response here so the diff --git a/messagefoundry/pipeline/wiring_runner.py b/messagefoundry/pipeline/wiring_runner.py index 5b4e6bbf..9124fee1 100644 --- a/messagefoundry/pipeline/wiring_runner.py +++ b/messagefoundry/pipeline/wiring_runner.py @@ -1815,7 +1815,7 @@ async def start_outbound(self, name: str) -> None: so it can't race a concurrent reload/stop (review M-10). Sharded: owner-only (ADR 0073).""" async with self._reload_lock: self._require_owned_destination(name) - self._start_outbound_unsafe(name) + await self._start_outbound_unsafe(name) async def stop_outbound(self, name: str) -> None: """PAUSE delivery on one outbound connection while RETAINING its queued rows PENDING (the @@ -1836,7 +1836,7 @@ async def restart_outbound(self, name: str) -> None: async with self._reload_lock: self._require_owned_destination(name) self._stop_outbound_unsafe(name) - self._start_outbound_unsafe(name) + await self._start_outbound_unsafe(name) def _stop_outbound_unsafe(self, name: str) -> None: """stop_outbound body without the reload lock (callers hold it). Sync + returns fast: it flags @@ -1934,7 +1934,7 @@ def _unpark_outbound_lane(self, name: str) -> None: else: self._outbound_resume.setdefault(name, asyncio.Event()).set() - def _ensure_destination_built(self, name: str) -> None: + async def _ensure_destination_built(self, name: str) -> None: """Build ``name``'s connector into ``_destinations`` if the lane has none β€” the missing half of an operator start (#115). The ``auto_start=False`` boot gate leaves a CONNECTOR-LESS lane, so a resume alone could never deliver a single byte: the advertised ``POST /connections/{name}/start`` @@ -1961,6 +1961,7 @@ def _ensure_destination_built(self, name: str) -> None: oc = self.registry.outbound.get(name) if oc is None or not oc.deployed: return + connector: DestinationConnector | None = None try: dest = _dest_config(oc, self._env_values, self._trust_anchor_policy, self._egress) check_egress_allowed(dest, self._egress) # fail-closed egress allowlist (WP-11c) @@ -1969,14 +1970,30 @@ def _ensure_destination_built(self, name: str) -> None: # (fail-closed/no-op default) posture. with active_hop_posture(self._hop_posture): connector = build_destination(dest) + # Opt-in at-start directory validation (#114) β€” the same hook _start_outbound awaits, so an + # operator START of a File/RemoteFile outbound with validate_directory=true gets the same + # refusal the engine start path gives it, isolated the same way (this method never raises). + await connector.validate_startup() except Exception as exc: + await self._aclose_quietly(connector, name) self._destinations.pop(name, None) self._record_failed(name, exc, kind="outbound") return self._destinations[name] = connector self._failed.pop(name, None) - def _start_outbound_unsafe(self, name: str) -> None: + async def _aclose_quietly(self, connector: DestinationConnector | None, name: str) -> None: + """Release whatever a BUILT-but-rejected connector allocated (a File alternate-credential worker + thread + token, ADR 0132; an HTTP client) β€” the lane is not going live, so it must not outlive + the failure. A close error is logged, never allowed to mask the failure that caused it.""" + if connector is None: + return + try: + await connector.aclose() + except Exception: + log.exception("closing the rejected connector for outbound %r raised", name) + + async def _start_outbound_unsafe(self, name: str) -> None: """start_outbound body without the reload lock. RESUMES delivery for a paused outbound, BUILDING its connector first if the lane has none (a start-disabled / DR-parked / failed lane is connector-less β€” see :meth:`_ensure_destination_built`). A connector that IS live is kept WARM (a @@ -1989,7 +2006,7 @@ def _start_outbound_unsafe(self, name: str) -> None: self._validate_outbound(name) if not self._deployed(name, "outbound"): raise NotDeployedError(name) - self._ensure_destination_built(name) + await self._ensure_destination_built(name) self._outbound_paused.discard(name) # The OPERATOR now owns this lane's UP state β€” the engine park (if any) is spent, and a reload # must respect the start (#115): drop the marker so _unpark_outbound_lane can't re-park it. @@ -2310,7 +2327,7 @@ def _record_failed(self, name: str, exc: BaseException, *, kind: str) -> None: except Exception: log.exception("alert sink raised on connection_stopped for %r", name) - def _start_outbound(self, name: str, oc: OutboundConnection) -> None: + async def _start_outbound(self, name: str, oc: OutboundConnection) -> None: """Build one outbound connector + spawn its delivery worker. A build failure (unresolvable ``env()`` / cert, an egress-allowlist refusal, a capture/backend mismatch) is ISOLATED (ADR 0031): the connection is recorded failed and the worker is STILL spawned, but with no @@ -2378,6 +2395,7 @@ def _start_outbound(self, name: str, oc: OutboundConnection) -> None: self._filtered.pop( name, None ) # at/above threshold this run β€” clear any prior parked marker + connector: DestinationConnector | None = None try: dest = _dest_config(oc, self._env_values, self._trust_anchor_policy, self._egress) check_egress_allowed(dest, self._egress) # fail-closed egress allowlist (WP-11c) @@ -2401,7 +2419,18 @@ def _start_outbound(self, name: str, oc: OutboundConnection) -> None: "support request/response capture (ADR 0013) β€” SQLite, Postgres, and SQL Server " "all do" ) + # Opt-in at-start directory validation (#114) β€” the outbound mirror of the source hook + # awaited in _start_inbound_unsafe. A File/RemoteFile outbound with validate_directory=true + # fails-fast HERE on a missing/unusable target directory; the default is a no-op on every + # connector, so every lane authored today starts byte-identically. Deliberately INSIDE this + # try: a DestinationStartupError then takes the SAME ADR-0031 isolation path as a build + # failure β€” the lane is recorded failed with NO connector and its worker is STILL spawned, + # so rows routed to it are retried + buildup-alerted, never dropped. On an outbound that + # degraded-lane state IS "invalid means not-started". Placed at start, NOT build_check: an + # intermittently-available directory must still let the graph BUILD. + await connector.validate_startup() except Exception as exc: + await self._aclose_quietly(connector, name) self._destinations.pop(name, None) # no live connector for a failed lane self._record_failed(name, exc, kind="outbound") self._spawn_worker(name) # drainsβ†’retries routed rows via the connector-None path @@ -2451,7 +2480,7 @@ async def start(self) -> None: # stays a backstop for genuinely fatal, graph-wide startup errors (the store, the # lookup executor), which still unwind + raise. for name, oc in self.registry.outbound.items(): - self._start_outbound(name, oc) + await self._start_outbound(name, oc) # Build the live-lookup executor from the graph (env-resolved + egress-checked here); # None when no DatabaseLookup is declared, keeping the transform path byte-identical. A # failure here is graph-wide (not one connection), so let it hit the backstop below. diff --git a/messagefoundry/transports/base.py b/messagefoundry/transports/base.py index 979a00fa..2478fc2c 100644 --- a/messagefoundry/transports/base.py +++ b/messagefoundry/transports/base.py @@ -270,6 +270,17 @@ class SourceStartupError(Exception): that logs-and-retries every poll). Distinct from :class:`DeliveryError` (a delivery outcome).""" +class DestinationStartupError(Exception): + """The outbound mirror of :class:`SourceStartupError`: a destination's **opt-in at-start + validation** (:meth:`DestinationConnector.validate_startup`) failed β€” a File/RemoteFile outbound + whose ``validate_directory`` toggle is on, pointed at a target directory that is **missing or + unusable** at start (BACKLOG #114). Raised from the runner's outbound start path so ADR 0031 + isolates the lane as ``failed`` (no live connector, its delivery worker still spawned so routed rows + are retried, never dropped) instead of the default run-time deferral β€” which on an outbound means + the target directory is silently ``mkdir``ed on the first write. Distinct from + :class:`DeliveryError` so a start-time refusal is never mistaken for a retryable send failure.""" + + def encode_wire_body(payload: str, encoding: str, *, transport: str) -> bytes: """Encode an outbound payload to the bytes that go on the wire (or to disk), failing CONTENT-FREE. @@ -484,6 +495,22 @@ async def send( async def aclose(self) -> None: return None + async def validate_startup(self) -> None: + """Optional **opt-in** at-start validity check for the destination's external resource, run by + the runner in ``_start_outbound`` right after the connector is built (BACKLOG #114) β€” the + outbound mirror of :meth:`SourceConnector.validate_startup`. The **default is a no-op**, so + every connector that does not override it (and every File/RemoteFile outbound that leaves + ``validate_directory`` off) is byte-identical: validation stays deferred to run time. The FILE + and REMOTEFILE destinations override it to **fail-fast** when ``validate_directory`` is on β€” a + missing/unusable target directory raises :class:`DestinationStartupError`, which the runner + isolates as an ADR-0031 ``failed`` lane. + + Like the source hook, this must **not** create anything β€” a merely-missing directory has to + FAIL. That is the whole reason it exists on an outbound: :meth:`test_connection` (the on-demand + ``POST /connections/{name}/test`` probe) **creates** the target directory on both FILE and + REMOTEFILE, so asking it whether the directory exists makes the answer yes.""" + return None + async def test_connection(self) -> None: """Probe the downstream peer for reachability β€” sending NO real payload and writing no message β€” for the ``POST /connections/{name}/test`` API. Returns on success; raises diff --git a/messagefoundry/transports/file.py b/messagefoundry/transports/file.py index d4c7b88e..d7db1743 100644 --- a/messagefoundry/transports/file.py +++ b/messagefoundry/transports/file.py @@ -43,6 +43,7 @@ from messagefoundry.transports.base import ( DeliveryError, DestinationConnector, + DestinationStartupError, InboundHandler, SourceConnector, SourceStartupError, @@ -206,6 +207,11 @@ def __init__(self, config: Destination) -> None: if "directory" not in s: raise ValueError("file destination requires a 'directory' setting") self.directory = Path(s["directory"]) + # Opt-in at-start directory validation (#114, ADR 0031 amendment). Default off = the historical + # run-time deferral, which on an outbound means the target dir is created on the first write. + # When on, the directory must already exist at start AND `_write` never creates it β€” an operator + # who said "never invent this path" gets that at delivery time too, not only at start. + self.validate_directory: bool = bool(s.get("validate_directory", False)) self.filename_template: str = s.get("filename", "{MSH-10}.hl7") # When two messages resolve to the same name, append a counter rather than clobber. self._overwrite: bool = bool(s.get("overwrite", False)) @@ -234,9 +240,34 @@ async def send( except OSError as exc: raise DeliveryError(f"file write failed: {exc}") from exc + async def validate_startup(self) -> None: + """Opt-in at-start directory validation (#114) β€” the outbound mirror of + :meth:`FileSource.validate_startup`. No-op unless ``validate_directory`` is set; then the target + directory must already exist (no mkdir β€” a merely-missing dir FAILS) and accept a write, since a + delivery writes there. A failure raises :class:`DestinationStartupError` so the runner isolates + the lane as ADR-0031 ``failed``: no live connector, the delivery worker still spawned, routed + rows retried rather than delivered into a directory the engine invented. + + The probe runs under the alternate credential when one is configured (#111), so it validates the + share under the same identity the delivery will use.""" + if not self.validate_directory: + return + try: + await self._run_fs(_probe_dir_startup, self.directory, require_write=True) + except OSError as exc: + raise DestinationStartupError( + f"file destination directory {self.directory} failed startup validation: {exc}" + ) from exc + async def test_connection(self) -> None: + # Under validate_directory the probe must NOT create either: otherwise POST + # /connections/{name}/test would silently repair the very typo the toggle exists to catch, and + # the next restart would then validate clean with nobody the wiser. try: - await self._run_fs(_probe_dir_writable, self.directory) + if self.validate_directory: + await self._run_fs(_probe_dir_startup, self.directory, require_write=True) + else: + await self._run_fs(_probe_dir_writable, self.directory) except OSError as exc: raise DeliveryError(f"file directory {self.directory} not writable: {exc}") from exc @@ -246,8 +277,42 @@ async def aclose(self) -> None: if self._cred_ctx is not None: await self._cred_ctx.close() + def _ensure_directory(self) -> None: + """Make the target directory usable for this write β€” and make a CREATION observable (#114). + + Default (``validate_directory`` off): the unchanged create-if-missing, except that a directory + this call actually created now logs a WARNING naming it. That silence is the defect: a typo'd + ``directory`` would otherwise be created on the first delivery and every message counted and + logged as delivered β€” because it was β€” into a path nobody is watching, with no error anywhere. + + ``validate_directory`` on: never create. The directory was validated at start; if it has since + vanished the write raises, and ``send`` maps that to a retryable :class:`DeliveryError`, so the + lane backs off and self-heals when the share returns instead of fabricating a local directory at + the mount point and delivering into it. + + The syscall count on the default path is unchanged: ``mkdir(parents=True, exist_ok=True)`` + already probed ``is_dir()`` on its ``FileExistsError`` branch, which is the common one.""" + if self.validate_directory: + if not self.directory.is_dir(): + raise FileNotFoundError( + f"destination directory {self.directory} does not exist and validate_directory is " + "on, so it is never created on write" + ) + return + try: + self.directory.mkdir(parents=True) + except FileExistsError: + if not self.directory.is_dir(): + raise # a non-directory sits at the configured path β€” the same OSError as before + else: + logger.warning( + "file destination CREATED missing directory %s β€” this delivery is landing in a " + "directory the engine just made; verify the configured path is the intended one", + self.directory, + ) + def _write(self, payload: str) -> None: - self.directory.mkdir(parents=True, exist_ok=True) + self._ensure_directory() name = render_filename(self.filename_template, payload, fallback="message.hl7") if self.compress == "gzip" and not name.endswith(".gz"): # Signal the on-disk format so a downstream reader (or a gunzip source) knows to unpack. diff --git a/messagefoundry/transports/remotefile.py b/messagefoundry/transports/remotefile.py index bf68157a..913caea2 100644 --- a/messagefoundry/transports/remotefile.py +++ b/messagefoundry/transports/remotefile.py @@ -67,6 +67,7 @@ from messagefoundry.transports.base import ( DeliveryError, DestinationConnector, + DestinationStartupError, InboundHandler, NegativeAckError, SourceConnector, @@ -147,8 +148,11 @@ def remove(self, path: str) -> None: """Delete the file at ``path``.""" @abc.abstractmethod - def ensure_dir(self, remote_dir: str) -> None: - """Best-effort create ``remote_dir`` (ignore "already exists").""" + def ensure_dir(self, remote_dir: str) -> bool: + """Best-effort create ``remote_dir`` (ignore "already exists"). Returns **True only when THIS + call created it** (#114) β€” the destination logs that, so a delivery landing in a directory the + engine just invented is distinguishable from a normal one. A best-effort failure (no permission, + a racing creator) returns False: nothing was created here.""" def _ftps_ssl_context( @@ -304,14 +308,17 @@ def rename(self, src: str, dst: str) -> None: def remove(self, path: str) -> None: self._op(lambda ftp: ftp.delete(path)) - def ensure_dir(self, remote_dir: str) -> None: - def run(ftp: ftplib.FTP) -> None: - try: # noqa: SIM105 + def ensure_dir(self, remote_dir: str) -> bool: + def run(ftp: ftplib.FTP) -> bool: + try: ftp.mkd(remote_dir) except ftplib.error_perm: - pass # already exists (or no permission) β€” best-effort, like File's mkdir(exist_ok) + # already exists (or no permission) β€” best-effort, like File's mkdir(exist_ok); either + # way THIS call did not create it, so the caller must not report a creation. + return False + return True - self._op(run) + return self._op(run) def _op(self, fn: Callable[[ftplib.FTP], _T]) -> _T: """Connect, run ``fn(ftp)``, always close. Maps ``ftplib`` failures to :class:`_RemoteError`: @@ -450,17 +457,19 @@ def rename(self, src: str, dst: str) -> None: def remove(self, path: str) -> None: self._op(lambda sftp: sftp.remove(path)) - def ensure_dir(self, remote_dir: str) -> None: - def run(sftp: Any) -> None: + def ensure_dir(self, remote_dir: str) -> bool: + def run(sftp: Any) -> bool: try: sftp.stat(remote_dir) except FileNotFoundError: - try: # noqa: SIM105 + try: sftp.mkdir(remote_dir) except OSError: - pass # racing creator / no permission β€” best-effort + return False # racing creator / no permission β€” best-effort + return True + return False # already there β€” nothing created - self._op(run) + return self._op(run) def _op(self, fn: Callable[[Any], _T]) -> _T: """Connect, open an SFTP channel, run ``fn(sftp)``, always close. Maps a host-key rejection @@ -631,6 +640,10 @@ def __init__(self, config: Destination) -> None: self._filename_template = str(s.get("filename", "{MSH-10}.hl7")) self._overwrite = bool(s.get("overwrite", False)) self._encoding: str = s.get("encoding", "utf-8") + # Opt-in at-start directory validation (#114, ADR 0031 amendment). Default off = the historical + # run-time deferral (ensure_dir creates the upload dir on the first send). When on, remote_dir + # must be listable at start AND _upload never creates it. + self._validate_directory: bool = bool(s.get("validate_directory", False)) async def send( self, payload: str, *, metadata: Mapping[str, str] | None = None @@ -651,10 +664,60 @@ async def send( ) from exc raise DeliveryError(str(exc)) from exc + async def validate_startup(self) -> None: + """Opt-in at-start directory validation (#114) β€” the outbound mirror of + :meth:`RemoteFileSource.validate_startup`. No-op unless ``validate_directory`` is set; then + ``remote_dir`` must be reachable and listable now. A listing is the only **no-create** probe + this client contract has (``ensure_dir`` creates, which is exactly what must not happen here), + so a no-such-dir / connect / auth failure raises :class:`DestinationStartupError` and the runner + isolates the lane as ADR-0031 ``failed``. Listing proves reachability and existence, not + writability β€” a share that lists but refuses a write still fails at the first delivery, where it + is retried and never dropped.""" + if not self._validate_directory: + return + try: + await asyncio.to_thread(self._client.list_dir, self._remote_dir) + except _RemoteError as exc: + raise DestinationStartupError( + f"REMOTEFILE destination directory {_redact(self._host, self._remote_dir)} failed " + f"startup validation: {exc}" + ) from exc + + def _prepare_remote_dir(self) -> None: + """Make ``remote_dir`` usable for this upload β€” and make a CREATION observable (#114). + + Default (``validate_directory`` off): the unchanged ``ensure_dir`` create-if-missing, except + that a directory this call actually created now logs a WARNING. That silence is the defect: a + typo'd ``remote_dir`` would otherwise be created on the partner's server and every message + counted and logged as delivered β€” because it was β€” into a path nobody is watching. + + ``validate_directory`` on: never create. The directory was validated at start, so a LIST is the + pre-flight check and its failure is re-raised as **transient**, which ``send`` maps to a + retryable :class:`DeliveryError`. The reclassification is the point: an SFTP/FTP no-such-dir is + a **permanent** error, so letting the upload fail on its own would dead-letter live traffic over + a share that is merely unmounted. It costs one extra round trip per delivery, on the opt-in + path only.""" + if self._validate_directory: + try: + self._client.list_dir(self._remote_dir) + except _RemoteError as exc: + raise _RemoteError( + f"REMOTEFILE upload directory {_redact(self._host, self._remote_dir)} is not " + f"available, and validate_directory is on so it is never created on send: {exc}", + permanent=False, + ) from exc + return + if self._client.ensure_dir(self._remote_dir): + logger.warning( + "REMOTEFILE destination CREATED missing directory %s β€” this delivery is landing in a " + "directory the engine just made; verify the configured remote_dir is the intended one", + _redact(self._host, self._remote_dir), + ) + def _upload(self, payload: str) -> None: name = render_filename(self._filename_template, payload, fallback="message.hl7") data = payload.encode(self._encoding) - self._client.ensure_dir(self._remote_dir) + self._prepare_remote_dir() final = posixpath.join(self._remote_dir, name) if not self._overwrite: final = self._unique(final) @@ -693,9 +756,14 @@ def _unique(self, final: str) -> str: async def test_connection(self) -> None: # Connect + authenticate + ensure the upload dir (the destination's normal first step) β€” no - # message data written. A failure is mapped like send()'s. + # message data written. A failure is mapped like send()'s. Under validate_directory the probe + # LISTS instead of ensuring: "never invent this path" has to hold for the on-demand probe too, + # or POST /connections/{name}/test would silently repair the typo the toggle exists to catch. try: - await asyncio.to_thread(self._client.ensure_dir, self._remote_dir) + if self._validate_directory: + await asyncio.to_thread(self._client.list_dir, self._remote_dir) + else: + await asyncio.to_thread(self._client.ensure_dir, self._remote_dir) except _RemoteError as exc: if exc.permanent: raise NegativeAckError( diff --git a/tests/test_connections_file.py b/tests/test_connections_file.py index 60d5e070..70a68f60 100644 --- a/tests/test_connections_file.py +++ b/tests/test_connections_file.py @@ -303,10 +303,11 @@ def test_duplicate_name_across_file_and_code_fails(tmp_path: Path) -> None: load_config(cfg) -def test_outbound_validate_directory_is_rejected_from_toml(tmp_path: Path) -> None: - # #114: the guard lives in build_outbound_connection, the single choke point BOTH authoring - # surfaces pass through β€” so the data-authored outbound is rejected by the same rule as the - # code-first one, with no second check in the TOML reader. This test pins that claim. +def test_outbound_validate_directory_loads_from_toml(tmp_path: Path) -> None: + # #114: the data-authored outbound reaches the connector through the same build_outbound_connection + # choke point as the code-first one, with no second check in the TOML reader β€” so the startup + # validation toggle is available on both authoring surfaces. (It was a WiringError while only the + # inbound half existed; both halves are built now.) cfg = _config( tmp_path, """ @@ -318,8 +319,8 @@ def test_outbound_validate_directory_is_rejected_from_toml(tmp_path: Path) -> No validate_directory = true """, ) - with pytest.raises(WiringError, match="validate_directory is an inbound-only option"): - load_config(cfg) + reg = load_config(cfg) + assert reg.outbound["OB_FILE"].spec.settings["validate_directory"] is True def test_unknown_transport_fails(tmp_path: Path) -> None: diff --git a/tests/test_remotefile_transport.py b/tests/test_remotefile_transport.py index a4c85740..de1c401d 100644 --- a/tests/test_remotefile_transport.py +++ b/tests/test_remotefile_transport.py @@ -29,7 +29,11 @@ from messagefoundry.config.wiring import Ftp, Sftp, WiringError from messagefoundry.pipeline.wiring_runner import check_egress_allowed, check_source_allowed from messagefoundry.transports import build_destination, build_source, remotefile -from messagefoundry.transports.base import DeliveryError, NegativeAckError +from messagefoundry.transports.base import ( + DeliveryError, + DestinationStartupError, + NegativeAckError, +) from messagefoundry.transports.remotefile import ( RemoteFileDestination, RemoteFileSource, @@ -55,16 +59,21 @@ def __init__( store_exc: _RemoteError | None = None, rename_exc: _RemoteError | None = None, retrieve_exc: _RemoteError | None = None, + list_exc: _RemoteError | None = None, ) -> None: self.files: dict[str, bytes] = dict(files or {}) self._sizes = sizes or {} self.ops: list[tuple[str, str]] = [] # (op, path) self.dirs: list[str] = [] + self._existing_dirs: set[str] = set() # #114: which dirs ensure_dir has already created self._store_exc = store_exc self._rename_exc = rename_exc self._retrieve_exc = retrieve_exc + self._list_exc = list_exc # #114: an unreachable/missing remote_dir def list_dir(self, remote_dir: str) -> list[tuple[str, int]]: + if self._list_exc is not None: + raise self._list_exc out: list[tuple[str, int]] = [] for path, data in self.files.items(): if posixpath.dirname(path) == remote_dir: @@ -94,8 +103,14 @@ def remove(self, path: str) -> None: self.ops.append(("remove", path)) self.files.pop(path, None) - def ensure_dir(self, remote_dir: str) -> None: + def ensure_dir(self, remote_dir: str) -> bool: + # #114: the contract now reports whether THIS call created the directory, so the caller can log + # a delivery that landed in a directory the engine just invented. self.dirs.append(remote_dir) + if remote_dir in self._existing_dirs: + return False + self._existing_dirs.add(remote_dir) + return True def _install_client(monkeypatch: pytest.MonkeyPatch, client: _FakeClient) -> None: @@ -1032,3 +1047,100 @@ def test_ftps_encrypted_client_key_wrong_password_raises(tmp_path: Path) -> None _ftps_ssl_context( {"host": "h", "tls_cert_file": cert, "tls_key_file": key, "tls_key_password": "WRONG"} ) + + +# --- #114 opt-in startup directory validation: the OUTBOUND half ------------- + +_UPLOAD_BODY = "MSH|^~\\&|A|B|C|D|20260810||ADT^A01|MSGX|P|2.5" + + +async def test_remote_destination_test_probe_creates_the_directory( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The REMOTEFILE half of the same measured claim as the File sibling: the on-demand probe ENSURES + # (creates) remote_dir, so it cannot answer the question a startup-validation toggle asks. + client = _FakeClient() + await _dest(monkeypatch, client).test_connection() + assert client.dirs == ["/in"] # ensure_dir, not a listing β€” the probe creates + + +async def test_remote_destination_validate_directory_off_is_noop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The item's own trigger: an intermittently-available remote directory (a listing fails right now) + # must NOT fail startup with the toggle off. Default = defer to run time, exactly as before. + client = _FakeClient(list_exc=_RemoteError("no such dir", permanent=True)) + await _dest(monkeypatch, client).validate_startup() # no raise + assert client.dirs == [] # and the hook created nothing + + +async def test_remote_destination_intermittent_dir_starts_and_then_delivers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The item's trigger end to end, on the default (lenient) setting: remote_dir is unreachable at + # start, so startup validation must NOT refuse the lane β€” and once the share comes back the upload + # goes through. This is why the toggle is opt-in rather than the default. + client = _FakeClient(list_exc=_RemoteError("share is down", permanent=False)) + dest = _dest(monkeypatch, client, filename="msg.hl7") + await dest.validate_startup() # start is not blocked by a share that is down right now + client._list_exc = None # the mount returns + await dest.send(_UPLOAD_BODY) + assert client.files["/in/msg.hl7"] == _UPLOAD_BODY.encode("utf-8") + + +async def test_remote_destination_validate_directory_refuses_unreachable_dir( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _FakeClient(list_exc=_RemoteError("no such dir", permanent=True)) + dest = _dest(monkeypatch, client, validate_directory=True) + with pytest.raises(DestinationStartupError): + await dest.validate_startup() + assert client.dirs == [] # LIST is the no-create probe β€” ensure_dir is never called + + +async def test_remote_destination_validate_directory_passes_when_listable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _FakeClient() + await _dest(monkeypatch, client, validate_directory=True).validate_startup() + assert client.dirs == [] + + +async def test_remote_destination_created_directory_is_logged( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + # Default arm, unchanged except that a CREATED upload directory is now loud. + client = _FakeClient() + dest = _dest(monkeypatch, client) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + await dest.send(_UPLOAD_BODY) + assert "CREATED missing directory" in caplog.text + caplog.clear() + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + await dest.send(_UPLOAD_BODY) + assert "CREATED missing directory" not in caplog.text # only a real creation is loud + + +async def test_remote_destination_validate_directory_upload_never_creates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Under the toggle the upload directory is never created at delivery time either, and the failure is + # deliberately RECLASSIFIED as transient: an SFTP/FTP no-such-dir is a PERMANENT error, so letting + # the upload fail naturally would dead-letter live traffic over a merely-unmounted share. + client = _FakeClient(list_exc=_RemoteError("no such dir", permanent=True)) + dest = _dest(monkeypatch, client, validate_directory=True) + with pytest.raises(DeliveryError) as exc: + await dest.send(_UPLOAD_BODY) + assert not isinstance(exc.value, NegativeAckError) # retried, never dead-lettered + assert client.dirs == [] # never created + assert client.ops == [] # and nothing was stored + + +async def test_remote_destination_validate_directory_test_probe_never_creates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _FakeClient(list_exc=_RemoteError("no such dir", permanent=True)) + dest = _dest(monkeypatch, client, validate_directory=True) + with pytest.raises(DeliveryError): + await dest.test_connection() + assert client.dirs == [] diff --git a/tests/test_startup_fault_isolation.py b/tests/test_startup_fault_isolation.py index 35d54826..2571aea9 100644 --- a/tests/test_startup_fault_isolation.py +++ b/tests/test_startup_fault_isolation.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio +import logging import socket from pathlib import Path @@ -287,6 +288,77 @@ async def test_file_validate_directory_off_defers_missing_dir( await runner.stop() +def _file_outbound_validate(outdir: Path, *, validate: bool) -> OutboundConnection: + settings: dict[str, object] = {"directory": str(outdir), "filename": "{MSH-10}.hl7"} + if validate: + settings["validate_directory"] = True + return OutboundConnection( + "file_out", + ConnectionSpec(ConnectorType.FILE, settings), + retry=RetryPolicy(backoff_seconds=0.05), + ) + + +async def test_file_outbound_validate_directory_isolates_missing_dir( + store: MessageStore, tmp_path: Path +) -> None: + # #114 remainder: an OUTBOUND File connection with validate_directory=true, pointed at a typo'd + # (missing) directory, is REFUSED at start β€” the lane is reported `failed` with no live connector + # while the engine stays up, and the directory is never fabricated. "Invalid means not-started" on + # an outbound IS the ADR-0031 degraded-lane state: the delivery worker still runs, so a message + # routed there is RETAINED pending and retried rather than delivered into an invented path. + inbox, missing = tmp_path / "in", tmp_path / "typo" + inbox.mkdir() + reg = Registry() + reg.add_inbound(_file_inbound(inbox)) + reg.add_outbound(_file_outbound_validate(missing, validate=True)) + reg.add_router("r", lambda m: ["h"]) + reg.add_handler("h", lambda m: Send("file_out", m)) + sink = _RecordingAlertSink() + runner = RegistryRunner(reg, store, poll_interval=0.02, alert_sink=sink) + await runner.start() + try: + assert runner.running # isolated, not fatal + assert "file_out" in runner.degraded_connections() + assert "DestinationStartupError" in (runner.connection_failed("file_out") or "") + assert "file_out" not in runner._destinations # no live connector + assert sink.stopped and sink.stopped[0][0] == "file_out" # alerted at start + assert not missing.exists() # the no-mkdir probe never fabricated it + (inbox / "a.hl7").write_bytes(ADT.encode("utf-8")) + await _wait_pending(store, "file_out") # retained + retried, never dropped + assert not missing.exists() # and nothing was written into an invented directory + finally: + await runner.stop() + + +async def test_file_outbound_validate_directory_off_defers_missing_dir( + store: MessageStore, tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + # The default, and the item's ACTUAL trigger: a target directory that is not there at start must NOT + # fail startup. The lane comes up clean and the first delivery creates the directory β€” unchanged + # behaviour β€” except that the creation now emits a WARNING naming the path, so it is + # distinguishable from a normal delivery. + inbox, outdir = tmp_path / "in", tmp_path / "late" + inbox.mkdir() + reg = Registry() + reg.add_inbound(_file_inbound(inbox)) + reg.add_outbound(_file_outbound_validate(outdir, validate=False)) + reg.add_router("r", lambda m: ["h"]) + reg.add_handler("h", lambda m: Send("file_out", m)) + runner = RegistryRunner(reg, store, poll_interval=0.02) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + await runner.start() + try: + assert runner.degraded_connections() == {} # validation deferred β€” the lane is clean + assert "file_out" in runner._destinations + (inbox / "a.hl7").write_bytes(ADT.encode("utf-8")) + await _until(lambda: (outdir / "MSG1.hl7").exists()) + await _wait_processed(store, "file_in") + finally: + await runner.stop() + assert "CREATED missing directory" in caplog.text + + async def test_valid_graph_starts_without_degradation(store: MessageStore, tmp_path: Path) -> None: # Regression: a fully-valid graph is unaffected β€” no degraded connections, and it delivers. inbox, outdir = tmp_path / "in", tmp_path / "out" diff --git a/tests/test_transports.py b/tests/test_transports.py index 8509495e..91ca0621 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -21,9 +21,15 @@ from messagefoundry.parsing.compression import gzip_compress, gzip_decompress from messagefoundry.parsing.peek import Peek from messagefoundry.transports import build_destination, build_source -from messagefoundry.transports.base import DeliveryError, NegativeAckError, SourceStartupError +from messagefoundry.transports.base import ( + DeliveryError, + DestinationStartupError, + NegativeAckError, + SourceStartupError, +) from messagefoundry.transports.file import ( DEFAULT_MAX_FILE_BYTES, + FileDestination, FileSource, _claim_unique, render_filename, @@ -678,6 +684,113 @@ async def test_file_validate_directory_leave_mode_requires_only_read( await move.validate_startup() # move mode needs write β†’ the failing probe raises +# --- #114 opt-in startup directory validation: the OUTBOUND half ------------- + + +def _file_dest(directory: Path, **over: object) -> FileDestination: + settings: dict[str, object] = {"directory": str(directory), "filename": "msg.hl7"} + settings.update(over) + dest = build_destination( + Destination(name="OB_FILE", type=ConnectorType.FILE, settings=settings) + ) + assert isinstance(dest, FileDestination) + return dest + + +async def test_file_destination_test_probe_creates_the_directory(tmp_path: Path) -> None: + # Pins WHY the outbound needs a startup hook of its own: #114's score rested on "a clean workaround + # via the on-demand test probe", and in this direction that premise is false. POST + # /connections/{name}/test CREATES the target directory, so the act of asking "does this directory + # exist?" makes the answer yes. Measured, not inferred β€” the rest of the design leans on it. + missing = tmp_path / "typo" + await _file_dest(missing).test_connection() + assert missing.is_dir() # the probe fabricated it + + +async def test_file_destination_validate_directory_off_is_noop(tmp_path: Path) -> None: + # Default: validate_startup is a no-op even on a missing directory. That is the item's own trigger + # (an intermittently-available target must NOT fail startup), so the toggle defaults to deferral. + missing = tmp_path / "nope" + await _file_dest(missing).validate_startup() # no raise + assert not missing.exists() # and the hook itself creates nothing + + +async def test_file_destination_validate_directory_refuses_missing_dir(tmp_path: Path) -> None: + # The opt-in arm: a typo'd target directory FAILS startup validation, and the no-mkdir probe never + # fabricates it (the same semantic gap the source hook closes β€” _probe_dir_writable would pass). + missing = tmp_path / "nope" + dest = _file_dest(missing, validate_directory=True) + with pytest.raises(DestinationStartupError): + await dest.validate_startup() + assert not missing.exists() + + +async def test_file_destination_validate_directory_passes_on_existing_writable_dir( + tmp_path: Path, +) -> None: + outdir = tmp_path / "out" + outdir.mkdir() + await _file_dest(outdir, validate_directory=True).validate_startup() # exists + writable + + +async def test_file_destination_created_directory_is_logged( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + # Default behaviour is unchanged β€” the directory is still created on write and the delivery still + # succeeds β€” but the creation is no longer SILENT. Without this line a typo'd directory would be + # created on first delivery and every message counted and logged as delivered (it was), into a path + # nobody is watching, with nothing anywhere saying so. + outdir = tmp_path / "made" + dest = _file_dest(outdir) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + await dest.send(ADT) + assert (outdir / "msg.hl7").exists() + assert "CREATED missing directory" in caplog.text + caplog.clear() + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + await dest.send(ADT) + assert "CREATED missing directory" not in caplog.text # only a real creation is loud + + +async def test_file_destination_non_directory_at_the_path_still_errors(tmp_path: Path) -> None: + # A regular FILE sitting at the configured directory path is still a mapped DeliveryError (retried, + # never a crash and never a silent success) after the create-detection rework swapped + # mkdir(exist_ok=True) for mkdir()+FileExistsError. + # + # Deliberately NOT claimed as a pin on that swap: sabotaging the re-raise (swallowing the + # FileExistsError) was measured and left this test GREEN, because the mkstemp two lines later fails + # with NotADirectoryError β€” another OSError β€” and maps to the same DeliveryError. That is precisely + # why the swap is safe, and why this asserts only the invariant that actually holds either way. + clash = tmp_path / "not_a_dir" + clash.write_text("i am a file", encoding="utf-8") + with pytest.raises(DeliveryError): + await _file_dest(clash).send(ADT) + assert clash.is_file() # untouched + + +async def test_file_destination_validate_directory_never_creates_on_write(tmp_path: Path) -> None: + # Under the toggle the target is not fabricated at DELIVERY time either: "this directory must + # exist" has to keep meaning that after start, or a share that vanished mid-run would be silently + # re-created at the mount point. The send fails RETRYABLY (DeliveryError), so the lane backs off and + # self-heals when the share returns β€” the row is never dropped. + missing = tmp_path / "gone" + dest = _file_dest(missing, validate_directory=True) + with pytest.raises(DeliveryError): + await dest.send(ADT) + assert not missing.exists() + + +async def test_file_destination_validate_directory_test_probe_never_creates(tmp_path: Path) -> None: + # ... and the on-demand probe honours the same rule under the toggle, so POST + # /connections/{name}/test cannot silently repair the typo the toggle exists to catch (after which + # the next restart would validate clean and the operator would never learn the path was wrong). + missing = tmp_path / "typo" + dest = _file_dest(missing, validate_directory=True) + with pytest.raises(DeliveryError): + await dest.test_connection() + assert not missing.exists() + + # --- #142 leave-in-place process-in-place disposition ----------------------- diff --git a/tests/test_wiring.py b/tests/test_wiring.py index e57337a1..d9f9d1cc 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -543,13 +543,12 @@ def test_streaming_knobs_accepted() -> None: assert ic.max_message_bytes == 100 * 1024 * 1024 -# --- #114: validate_directory is inbound-only --------------------------------- +# --- #114: validate_directory works in BOTH directions ------------------------ # -# File()/Sftp()/Ftp() are single factories serving BOTH directions, so they cannot reject the option -# themselves β€” only the bind knows the direction. Before this guard an outbound carrying -# validate_directory=True was accepted and silently ignored (no destination reads it, and -# DestinationConnector has no validate_startup hook), so an operator asking for fail-fast validation -# got none and saw no error. +# It was inbound-only for one reason β€” no destination read it, and DestinationConnector had no +# validate_startup hook β€” so an outbound carrying validate_directory=True was first silently ignored +# and then (2026-08-03) rejected outright. Both halves are built now, so the rejection is gone and the +# option is honoured on an outbound: the destination hook fails start on a missing target directory. def _spec_with_validate_directory(kind: str, directory: str) -> ConnectionSpec: @@ -561,11 +560,10 @@ def _spec_with_validate_directory(kind: str, directory: str) -> ConnectionSpec: @pytest.mark.parametrize("kind", ["file", "sftp", "ftp"]) -def test_outbound_validate_directory_is_rejected(kind: str, tmp_path: Path) -> None: +def test_outbound_validate_directory_now_builds(kind: str, tmp_path: Path) -> None: spec = _spec_with_validate_directory(kind, str(tmp_path / "out")) - with pytest.raises(WiringError, match="validate_directory is an inbound-only option") as exc: - build_outbound_connection("OB_X", spec) - assert "'OB_X'" in str(exc.value) # the message names the offending connection + oc = build_outbound_connection("OB_X", spec) + assert oc.spec.settings["validate_directory"] is True # reaches the connector, not rejected @pytest.mark.parametrize("kind", ["file", "sftp", "ftp"]) @@ -578,7 +576,7 @@ def test_inbound_validate_directory_still_builds(kind: str, tmp_path: Path) -> N @pytest.mark.parametrize("kind", ["file", "sftp", "ftp"]) def test_outbound_without_validate_directory_is_unaffected(kind: str, tmp_path: Path) -> None: - # The factories write validate_directory=False into settings unconditionally, so the guard has to + # The factories write validate_directory=False into settings unconditionally, so the toggle has to # be truthy-only β€” every outbound authored today must keep building byte-identically. directory = str(tmp_path / "out") spec = (