From 4da73c8a1cf8685d78d5a8644fd67de6a542a824 Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 19:25:41 -0500 Subject: [PATCH 1/6] X12FrameReader marks a read of its interchange's own line ends trailer_only (BACKLOG #2938) A newline after an X12 interchange's IEA terminator that arrived in a read of its own decoded nothing, so the shared frame clock started on it and a quiet, healthy peer would be cut as frame_deadline at max_frame_seconds. X12FrameReader now carries an eagerly computed trailer_only, set when feed() is called: true only for a read holding nothing but CR/LF bytes, at most two in all, directly after a completed interchange's full terminator. Any other byte ends the allowance, so a peer cannot trickle line ends to hold the deadline off. The rule mirrors FrameDecoder's, with local constants so the parsing layer imports nothing from framing.py. The flag only observes: the bytes still reach the buffer, so the 2-byte noise tail, the #1595 resume points, in_frame and the reuse path's leftover reading are unchanged. The X12 listener passes it to FrameClock.after_read. Tests: x12 rows (CR LF alone, LF after a CR sent with the IEA, and the controls Z and three line ends, which still cut) in the shared line-end test; the two non-control rows fail on the unfixed wiring. A reader unit test in tests/test_x12_parsing.py. --- messagefoundry/parsing/x12/interchange.py | 42 ++++++++++++++++ messagefoundry/transports/admission.py | 12 +++-- messagefoundry/transports/x12.py | 11 ++-- tests/test_listener_admission.py | 8 ++- tests/test_x12_parsing.py | 61 +++++++++++++++++++++++ 5 files changed, 124 insertions(+), 10 deletions(-) diff --git a/messagefoundry/parsing/x12/interchange.py b/messagefoundry/parsing/x12/interchange.py index 8c4283a6b4..1c55c803c5 100644 --- a/messagefoundry/parsing/x12/interchange.py +++ b/messagefoundry/parsing/x12/interchange.py @@ -43,6 +43,14 @@ # A segment that is the IEA trailer, after cosmetic whitespace. Matched in place with pos/endpos, so the # check never copies a segment, however long. _IEA_SEGMENT = re.compile(rb"[ \t\r\n\x0b\x0c]*IEA") +#: Line-end bytes a sender may put after an interchange's closing terminator, such as a newline after +#: ``IEA*...~``. Up to :data:`_FRAME_TAIL_MAX` of them directly after a completed interchange are its +#: ending, however reads split them (vault BACKLOG #2938). The cap keeps a peer from trickling them +#: to hold a listener's frame deadline off: past it they are inter-interchange noise like any byte. +#: The same rule as ``messagefoundry.framing.FrameDecoder``'s, kept local so this pure parsing module +#: imports nothing from the transport layer. +_FRAME_TAIL_BYTES = frozenset(b"\r\n") +_FRAME_TAIL_MAX = 2 def split(raw: str) -> list[str]: @@ -101,6 +109,15 @@ def __init__(self, max_interchange_bytes: int | None = None) -> None: # received begins; ``_scan`` is where the terminator search picks up, at or after it. self._seg_start = 0 self._scan = 0 + #: Line-end bytes the last completed interchange may still take as its ending; 0 once any + #: other byte has followed it. See :data:`_FRAME_TAIL_BYTES`. + self._tail_left = 0 + #: Whether the most recent :meth:`feed` held nothing but line-end bytes ending the + #: interchange before it. Such a read finishes that interchange rather than starting anything, + #: which is how a listener's frame deadline tells it from inter-interchange noise (vault + #: BACKLOG #2938). Set when :meth:`feed` is called, before its generator is iterated. It only + #: observes: the bytes still go to the buffer, so the noise handling is unchanged. + self.trailer_only = False @property def in_frame(self) -> bool: @@ -110,6 +127,18 @@ def in_frame(self) -> bool: return self._buf.startswith(b"ISA") def feed(self, data: bytes) -> Iterator[bytes]: + """Yield each interchange completed by ``data``, carrying any partial one to the next call. + + :attr:`trailer_only` is decided here, eagerly, so it never describes an earlier read; the + reassembly itself still runs lazily as the generator is iterated. + """ + owed = _leading_line_ends(data, self._tail_left) + self.trailer_only = 0 < owed == len(data) + # Any other byte ends the allowance: what follows it is not the interchange's ending. + self._tail_left = self._tail_left - owed if self.trailer_only else 0 + return self._frames(data) + + def _frames(self, data: bytes) -> Iterator[bytes]: self._buf.extend(data) while True: frame = self._take_one() @@ -163,6 +192,10 @@ def _take_one(self) -> bytes | None: frame = bytes(buf[:end]) del buf[:end] self._seg_start = self._scan = 0 + # What the read held past the terminator: line ends alone are owed to this + # interchange, so the allowance left is what they have not used yet. + owed = _leading_line_ends(buf, _FRAME_TAIL_MAX) + self._tail_left = _FRAME_TAIL_MAX - owed if owed == len(buf) else 0 return frame self._seg_start = self._scan = idx + len(terminator) @@ -177,6 +210,7 @@ def _terminator(buf: bytearray) -> bytes: def _clear(self) -> None: self._buf.clear() self._seg_start = self._scan = 0 + self._tail_left = 0 def _check_cap(self) -> None: # Called from every path that returns with the interchange still open. Only one of those can @@ -191,6 +225,14 @@ def _check_cap(self) -> None: ) +def _leading_line_ends(data: bytes | bytearray, limit: int) -> int: + """How many of the first ``limit`` bytes of ``data`` are line ends, up to the first that is not.""" + count, size = 0, min(limit, len(data)) + while count < size and data[count] in _FRAME_TAIL_BYTES: + count += 1 + return count + + def check_integrity(raw: str) -> list[str]: """Structural integrity tie-out for one interchange (a pure helper, opt-in). diff --git a/messagefoundry/transports/admission.py b/messagefoundry/transports/admission.py index c229595c8d..df2b0f6b46 100644 --- a/messagefoundry/transports/admission.py +++ b/messagefoundry/transports/admission.py @@ -357,11 +357,13 @@ def after_read(self, *, in_frame: bool, decoded: int, trailer_only: bool) -> Non ``in_frame`` would never reach a peer trickling them. So any bytes must lead to a completed frame within ``max_frame_seconds`` of read time; the engine's own waits are withheld. - ``trailer_only`` (:attr:`~messagefoundry.framing.FrameDecoder.trailer_only`) is the one - exception: a read holding only the line-end bytes that end the frame before it, such as - MLLP's CR arriving after its FS, belongs to that completed frame and starts nothing (vault - BACKLOG #2847). Without it a healthy peer that went quiet after such a read would be cut - here. It is required, so a new caller cannot leave it out by accident. + ``trailer_only`` (:attr:`~messagefoundry.framing.FrameDecoder.trailer_only`, or + :attr:`~messagefoundry.parsing.x12.interchange.X12FrameReader.trailer_only` for X12) is the + one exception: a read holding only the line-end bytes that end the frame before it, such as + MLLP's CR arriving after its FS or a newline after an X12 ``IEA``, belongs to that completed + frame and starts nothing (vault BACKLOG #2847, #2938). Without it a healthy peer that went + quiet after such a read would be cut here. It is required, so a new caller cannot leave it + out by accident. """ if decoded: self.opened_at = time.monotonic() if in_frame else None diff --git a/messagefoundry/transports/x12.py b/messagefoundry/transports/x12.py index b3140747d2..c71cd5db8d 100644 --- a/messagefoundry/transports/x12.py +++ b/messagefoundry/transports/x12.py @@ -851,10 +851,13 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW ) break # Reached on the success path alone: every arm above breaks or raises. - # X12FrameReader reports no end-of-frame bytes yet, so a newline after the IEA - # segment terminator that arrives in a read of its own starts the clock. That - # gap is open, not intended (vault BACKLOG #2847 review). - clock.after_read(in_frame=decoder.in_frame, decoded=decoded, trailer_only=False) + # A newline after the IEA segment terminator that arrives in a read of its own + # ends that interchange and starts no clock (vault BACKLOG #2938). + clock.after_read( + in_frame=decoder.in_frame, + decoded=decoded, + trailer_only=decoder.trailer_only, + ) except OSError as exc: failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) diff --git a/tests/test_listener_admission.py b/tests/test_listener_admission.py index c43ae346f4..bca48a3b9f 100644 --- a/tests/test_listener_admission.py +++ b/tests/test_listener_admission.py @@ -433,6 +433,11 @@ async def body(port: int) -> None: ("tcp", b"\r", False), ("mllp", b"Z", True), # CONTROL: any other byte after a completed frame starts the clock ("mllp", b"\r\n\r", True), # CONTROL: line ends past the two-byte allowance are noise + # vault BACKLOG #2938: the newline after an X12 IEA terminator takes the same allowance. + ("x12", b"\r\n", False), + ("x12", b"\n", False), # the CR arrived with the IEA, so the LF is the second byte + ("x12", b"Z", True), # CONTROL: a byte that is not a line end starts the clock + ("x12", b"\r\n\r", True), # CONTROL: three line ends are past the allowance ], ) async def test_line_end_bytes_ending_a_frame_do_not_start_the_frame_deadline( @@ -452,7 +457,8 @@ async def handler(raw: bytes) -> str | None: framing = {"framing": "mllp"} if kind == "tcp" else {} source = _build(kind, max_frame_seconds=0.3, receive_timeout=5.0, **framing) source.on_connection_event = events - whole = _whole_frame("mllp") + # X12's frame is the interchange through its IEA terminator; the CR here is its first line end. + whole = _whole_frame("x12") + b"\r" if kind == "x12" else _whole_frame("mllp") # The `\n` row sends the CR with the frame, so the LF is the second line-end byte. first = whole if after == b"\n" else whole[:-1] diff --git a/tests/test_x12_parsing.py b/tests/test_x12_parsing.py index 539f07d389..bd581d3ffd 100644 --- a/tests/test_x12_parsing.py +++ b/tests/test_x12_parsing.py @@ -671,6 +671,67 @@ def test_frame_reader_crlf_terminator_byte_at_a_time() -> None: assert out == [data] +def test_frame_reader_marks_only_an_interchanges_own_line_ends_as_trailer_only() -> None: + # vault BACKLOG #2938: up to two CR/LF bytes after a completed interchange are its ending, so a + # listener's frame deadline must not start on them. Mirrors the FrameDecoder test in + # tests/test_listener_admission.py. + data = interchange().encode("utf-8") + reader = X12FrameReader() + assert list(reader.feed(data)) == [data] + assert not reader.trailer_only, "a read that completed an interchange is not trailer-only" + assert list(reader.feed(b"\r")) == [] + assert reader.trailer_only, "the CR after the IEA terminator" + assert list(reader.feed(b"\n")) == [] + assert reader.trailer_only, "the LF of a CR LF ending" + # CONTROL: a third line-end byte is past the allowance, so it is noise like any other byte. + assert list(reader.feed(b"\r")) == [] + assert not reader.trailer_only + # CONTROL: a byte that is not a line end ends the allowance for what follows it too. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"Z")) == [] + assert not reader.trailer_only + assert list(reader.feed(b"\n")) == [] + assert not reader.trailer_only + # A line end that arrived with the interchange uses up part of the allowance. + assert list(reader.feed(data + b"\r")) == [data] + assert list(reader.feed(b"\n")) == [] + assert reader.trailer_only + assert list(reader.feed(b"\n")) == [] + assert not reader.trailer_only + # Both line ends in one read of their own. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"\r\n")) == [] + assert reader.trailer_only + # CONTROL: three in one read are past the allowance. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"\r\n\r")) == [] + assert not reader.trailer_only + # Decided when feed() is called, before its generator runs, so it never describes a past read. + assert list(reader.feed(data)) == [data] + reader.feed(b"Z") # not iterated + assert not reader.trailer_only + # Before any interchange there is nothing for a line end to end. + fresh = X12FrameReader() + assert list(fresh.feed(b"\r\n")) == [] + assert not fresh.trailer_only + # A CR+LF segment terminator is the interchange's own, so the allowance starts after it. + crlf = interchange(term="\r\n").encode("utf-8") + assert list(fresh.feed(crlf)) == [crlf] + assert list(fresh.feed(b"\r\n")) == [] + assert fresh.trailer_only + + +def test_frame_reader_trailer_bytes_still_reach_the_noise_buffer() -> None: + # The flag only observes. The line ends still sit in the 2-byte noise tail, so the leftover + # reading the reuse path takes from the buffer, and the next interchange's framing, are unchanged. + data = interchange().encode("utf-8") + reader = X12FrameReader() + assert list(reader.feed(data + b"\r\n")) == [data] + assert reader._buf == b"\r\n" + assert list(reader.feed(data)) == [data] + assert not reader.in_frame + + def test_integrity_group_and_transaction_mismatches() -> None: # GS06/GE02 control-number mismatch. assert any("GE02" in p for p in check_integrity(interchange().replace("GE*1*1~", "GE*1*9~"))) From 84159c4dd22798eb1c50b2b70a89ac671194f5ae Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 19:48:27 -0500 Subject: [PATCH 2/6] X12FrameReader's trailer check is a length-gated strip (BACKLOG #2938) A /simplify pass: both callers of the counting helper only asked whether every byte was a line end, so the helper is now that yes/no check. The length test runs first, so a large leftover buffer is never copied. The reset of the tail allowance in _clear() is dropped: no path reaches it with an allowance left, because a non-zero allowance means no interchange is open to overflow the cap. Behaviour is unchanged; the tests from the previous commit pin it. --- messagefoundry/parsing/x12/interchange.py | 22 +++++++++------------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/messagefoundry/parsing/x12/interchange.py b/messagefoundry/parsing/x12/interchange.py index 1c55c803c5..cb38dc5050 100644 --- a/messagefoundry/parsing/x12/interchange.py +++ b/messagefoundry/parsing/x12/interchange.py @@ -49,7 +49,7 @@ #: to hold a listener's frame deadline off: past it they are inter-interchange noise like any byte. #: The same rule as ``messagefoundry.framing.FrameDecoder``'s, kept local so this pure parsing module #: imports nothing from the transport layer. -_FRAME_TAIL_BYTES = frozenset(b"\r\n") +_FRAME_TAIL_BYTES = b"\r\n" _FRAME_TAIL_MAX = 2 @@ -132,10 +132,9 @@ def feed(self, data: bytes) -> Iterator[bytes]: :attr:`trailer_only` is decided here, eagerly, so it never describes an earlier read; the reassembly itself still runs lazily as the generator is iterated. """ - owed = _leading_line_ends(data, self._tail_left) - self.trailer_only = 0 < owed == len(data) + self.trailer_only = bool(data) and _only_line_ends(data, self._tail_left) # Any other byte ends the allowance: what follows it is not the interchange's ending. - self._tail_left = self._tail_left - owed if self.trailer_only else 0 + self._tail_left = self._tail_left - len(data) if self.trailer_only else 0 return self._frames(data) def _frames(self, data: bytes) -> Iterator[bytes]: @@ -194,8 +193,8 @@ def _take_one(self) -> bytes | None: self._seg_start = self._scan = 0 # What the read held past the terminator: line ends alone are owed to this # interchange, so the allowance left is what they have not used yet. - owed = _leading_line_ends(buf, _FRAME_TAIL_MAX) - self._tail_left = _FRAME_TAIL_MAX - owed if owed == len(buf) else 0 + tail = _only_line_ends(buf, _FRAME_TAIL_MAX) + self._tail_left = _FRAME_TAIL_MAX - len(buf) if tail else 0 return frame self._seg_start = self._scan = idx + len(terminator) @@ -210,7 +209,6 @@ def _terminator(buf: bytearray) -> bytes: def _clear(self) -> None: self._buf.clear() self._seg_start = self._scan = 0 - self._tail_left = 0 def _check_cap(self) -> None: # Called from every path that returns with the interchange still open. Only one of those can @@ -225,12 +223,10 @@ def _check_cap(self) -> None: ) -def _leading_line_ends(data: bytes | bytearray, limit: int) -> int: - """How many of the first ``limit`` bytes of ``data`` are line ends, up to the first that is not.""" - count, size = 0, min(limit, len(data)) - while count < size and data[count] in _FRAME_TAIL_BYTES: - count += 1 - return count +def _only_line_ends(data: bytes | bytearray, limit: int) -> bool: + """Whether ``data`` is at most ``limit`` bytes, every one a line end. The length is checked + first, so a large buffer is never copied by the strip.""" + return len(data) <= limit and not data.strip(_FRAME_TAIL_BYTES) def check_integrity(raw: str) -> list[str]: From 3a9e47087eceb11b4ad414b788a9c4a513b0f1c6 Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 19:51:29 -0500 Subject: [PATCH 3/6] A connection's release carries the epoch that admitted it, so a straggler cannot undercount a restarted listener's per-host cap (BACKLOG #2939) ListenerAdmission.admit() now returns an epoch, and release(peer_host, epoch) requires it, with no default, so mypy strict flags any call site that leaves it out. reset() advances the epoch only after its "not stopping" guard, so a late stop() that skips its reset leaves earlier releases counting, as they must: the tables still hold their slots. A release from an older epoch still gives back `active`, which reset() never clears (the socket was open and counted until then), and leaves per_host and host_capacity_warned alone. That answers the row's closing step 3 on `active`. All four listeners (MLLP, raw TCP, X12, HTTP) keep the token from admit and pass it to release; in http_listener.py only the admit and release lines in _on_client changed. The MLLPSource.stop() comment that called this "a separate change" now says what the epoch does. Tests: restart, refill the host to its cap, release the pre-restart token: the count is unchanged and check() still refuses (fails without the epoch check); and a skipped reset() followed by an old-token release still lowers the count (fails if the epoch advances before the guard). BACKLOG #2938 and BACKLOG #2939 ride this branch. Proposed PR title: X12 trailer newline no longer starts the frame deadline; a post-restart straggler no longer undercounts the per-host cap (BACKLOG #2938 #2939) Proposed banner, #2938: SHIPPED -- X12FrameReader exposes trailer_only, set eagerly in feed(): true only for a read of at most two CR/LF bytes directly after a completed interchange's full terminator; any other byte ends the allowance. The X12 listener passes it to FrameClock.after_read. x12 rows, with controls (a non-line-end byte, three line ends), added to the shared line-end test in tests/test_listener_admission.py; a reader unit test in tests/test_x12_parsing.py. framing.py untouched. Proposed banner, #2939: SHIPPED -- ListenerAdmission.admit returns an epoch, release requires it, reset advances it past its stopping guard. A release from an earlier epoch lowers `active` only (closing step 3: `active` counts open sockets across runs, so it still drops). All four listeners pass the token. Tests pin the restart-refill-stale-release case and the skipped-reset case. --- messagefoundry/transports/admission.py | 36 ++++++++--- messagefoundry/transports/http_listener.py | 4 +- messagefoundry/transports/mllp.py | 15 +++-- messagefoundry/transports/tcp.py | 4 +- messagefoundry/transports/x12.py | 4 +- tests/test_listener_admission.py | 70 +++++++++++++++++++--- 6 files changed, 102 insertions(+), 31 deletions(-) diff --git a/messagefoundry/transports/admission.py b/messagefoundry/transports/admission.py index df2b0f6b46..453648bb40 100644 --- a/messagefoundry/transports/admission.py +++ b/messagefoundry/transports/admission.py @@ -119,9 +119,9 @@ class Refusal: class ListenerAdmission: """The pre-read admission decision for one socket listener, and the counts it rests on. - Call :meth:`check` first. ``None`` means admit: call :meth:`admit`, and pair it with exactly one - :meth:`release` in a ``finally``. A :class:`Refusal` means close the socket unread; nothing was - taken, so nothing is released. + Call :meth:`check` first. ``None`` means admit: call :meth:`admit`, keep the epoch it returns, + and pair it with exactly one :meth:`release` of that epoch in a ``finally``. A :class:`Refusal` + means close the socket unread; nothing was taken, so nothing is released. ``max_connections`` counts sockets on this listener. ``max_connections_per_host`` counts sockets from one peer address, so one address cannot take the whole listener. It keys on the source IP, @@ -152,6 +152,10 @@ def __init__( #: Hosts already warned about their per-host cap this episode, cleared with the count above. self.host_capacity_warned: set[str] = set() self.refusal_log = RefusalLog() + #: Which run of the listener the per-host tables describe. :meth:`admit` hands it to the + #: connection and :meth:`reset` advances it, so a release from an earlier run can be told + #: from one the tables still count (vault BACKLOG #2939). + self._epoch = 0 def check(self, writer: asyncio.StreamWriter, peer_host: str | None) -> Refusal | None: """Decide whether to admit this connection. Logs a refusal, throttled; emits no event.""" @@ -173,17 +177,27 @@ def check(self, writer: asyncio.StreamWriter, peer_host: str | None) -> Refusal return _AT_HOST_CAPACITY return None - def admit(self, peer_host: str | None) -> None: - """Take one connection slot, globally and (when the peer has an address) for that host.""" + def admit(self, peer_host: str | None) -> int: + """Take one connection slot, globally and (when the peer has an address) for that host. + + Returns the epoch to hand back to :meth:`release`.""" self.active += 1 if peer_host is not None: self.per_host[peer_host] = self.per_host.get(peer_host, 0) + 1 + return self._epoch - def release(self, peer_host: str | None) -> None: + def release(self, peer_host: str | None, epoch: int) -> None: """Give both slots back. One place for both counters, so they cannot drift: a per-host count - left behind by a missed release would lock that peer out until restart.""" + left behind by a missed release would lock that peer out until restart. + + ``epoch`` is what :meth:`admit` returned, and it is required, so a call site cannot leave it + out. A release from an earlier epoch is a connection that outlived a stop: :meth:`reset` + already dropped its per-host slot, and the table now counts the restarted listener's + connections, so it touches neither that table nor the warned set (vault BACKLOG #2939). + It still gives back ``active``, which :meth:`reset` leaves alone: that socket was open, and + counted, until now, whichever run admitted it.""" self.active -= 1 - if peer_host is None: + if peer_host is None or epoch != self._epoch: return remaining = self.per_host.get(peer_host, 0) - 1 if remaining > 0: @@ -199,9 +213,13 @@ def reset(self) -> None: A no-op once the listener has started again: a stop() that finishes after a restart must not wipe the counts of connections the restarted listener has admitted. Call it from a - ``finally`` so a cancelled stop() still clears the tables.""" + ``finally`` so a cancelled stop() still clears the tables. + + Clearing advances the epoch, so a connection admitted before it that releases later gives + back no per-host slot it no longer holds.""" if not self.stopping: return + self._epoch += 1 self.per_host.clear() self.host_capacity_warned.clear() diff --git a/messagefoundry/transports/http_listener.py b/messagefoundry/transports/http_listener.py index d767c580e8..63723daec7 100644 --- a/messagefoundry/transports/http_listener.py +++ b/messagefoundry/transports/http_listener.py @@ -1078,7 +1078,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW ), ) return # pre-ingress refusal — nothing accepted, so nothing to count - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) established = True # set with the admit, so the finally below always releases it await self._emit_event("established", peer_host=peer_host) failed = await self._serve_one(reader, writer, peer_host=peer_host) @@ -1094,7 +1094,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW await self._write_safely(writer, build_response(500, '{"error":"internal error"}')) finally: if established: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) self._clients.discard(writer) if task is not None: self._client_tasks.discard(task) diff --git a/messagefoundry/transports/mllp.py b/messagefoundry/transports/mllp.py index 24ab2138bf..6cd5c7f0ce 100644 --- a/messagefoundry/transports/mllp.py +++ b/messagefoundry/transports/mllp.py @@ -2192,12 +2192,11 @@ async def stop(self) -> None: # at the next promotion) may not run its `finally` before the restart, so a count it # holds would survive into the restarted listener, and `release`'s own docstring names # what that costs: a per-host count left behind by a missed release locks that peer - # out. The limit, stated so it is not over-read: a straggler that releases AFTER the - # restart has re-admitted its host decrements the new count, so the per-host cap can - # undercount by one per such straggler while that host's newer connections stay open. - # Tying a release to the run - # that admitted it would close that, and is a separate change. `active` keeps its - # existing behaviour: the global cap that predates this row reads it. + # out. A straggler that releases AFTER the restart has re-admitted its host carries the + # epoch of the run that admitted it, and `reset` advances the epoch, so that release + # leaves the restarted listener's per-host count alone rather than undercounting it + # (vault BACKLOG #2939). `active` still drops on it: that socket was open until then, + # and the global cap that predates this row reads it. self._admission.reset() # Now that no client handlers are in flight, this should complete promptly — but on the Windows # ProactorEventLoop a still-pending overlapped accept/read can make wait_closed() never return, @@ -2374,7 +2373,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # refused (closed in the outer finally) - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) # The `established` emit sits INSIDE the try whose finally releases the slot. It is # fail-soft against `Exception` but not against `CancelledError`, and stop() cancels # straggling client tasks — so a cancellation delivered here used to escape past @@ -2500,7 +2499,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/messagefoundry/transports/tcp.py b/messagefoundry/transports/tcp.py index f77d439bb7..f596f30600 100644 --- a/messagefoundry/transports/tcp.py +++ b/messagefoundry/transports/tcp.py @@ -638,7 +638,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # closed in the outer finally - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) try: # Inside the try whose finally releases the slot, so a cancellation delivered at # this await cannot leak a per-host count. @@ -718,7 +718,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/messagefoundry/transports/x12.py b/messagefoundry/transports/x12.py index c71cd5db8d..61d8239e73 100644 --- a/messagefoundry/transports/x12.py +++ b/messagefoundry/transports/x12.py @@ -777,7 +777,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW if refusal.kind is not None: await self._emit_event(refusal.kind, peer_host=peer_host, reason=refusal.reason) return # closed in the outer finally - self._admission.admit(peer_host) + epoch = self._admission.admit(peer_host) try: # Inside the try whose finally releases the slot, so a cancellation delivered at # this await cannot leak a per-host count. @@ -862,7 +862,7 @@ async def _on_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamW failed = True # peer reset; nothing to do but drop the connection await self._emit_event("peer_reset", peer_host=peer_host, reason=safe_exc(exc)) finally: - self._admission.release(peer_host) + self._admission.release(peer_host, epoch) finally: self._clients.discard(writer) if task is not None: diff --git a/tests/test_listener_admission.py b/tests/test_listener_admission.py index bca48a3b9f..cd262599cd 100644 --- a/tests/test_listener_admission.py +++ b/tests/test_listener_admission.py @@ -603,21 +603,23 @@ def test_the_refusal_log_defaults_match_the_dicom_server() -> None: # --- Stopping, and the slot accounting ------------------------------------------------------------- +class _PeerWriter: + """The one writer method admission reads, for a peer at ``192.0.2.7``.""" + + def get_extra_info(self, name: str, default: object = None) -> object: + return ("192.0.2.7", 4000) if name == "peername" else default + + def test_admission_refuses_while_stopping_and_releases_what_it_admits() -> None: gate = ListenerAdmission( transport="TCP", max_connections=2, max_connections_per_host=1, source_ip_allowlist=None ) - - class _W: - def get_extra_info(self, name: str, default: object = None) -> object: - return ("192.0.2.7", 4000) if name == "peername" else default - - w: Any = _W() + w: Any = _PeerWriter() assert gate.check(w, "192.0.2.7") is None - gate.admit("192.0.2.7") + epoch = gate.admit("192.0.2.7") refusal = gate.check(w, "192.0.2.7") assert refusal is not None and refusal.reason == "max_connections_per_host" - gate.release("192.0.2.7") + gate.release("192.0.2.7", epoch) assert gate.active == 0 and gate.per_host == {} assert gate.host_capacity_warned == set() gate.stopping = True @@ -625,6 +627,58 @@ def get_extra_info(self, name: str, default: object = None) -> object: assert refusal is not None and refusal.kind is None, "a stop refusal emits no event" +def test_a_release_from_before_a_restart_leaves_the_restarted_count_alone() -> None: + """vault BACKLOG #2939: a connection that outlives its listener's stop() and releases after the + restart has re-admitted its host must not take a slot from the new count. Without the epoch, the + straggler's release would drop the host to one below the cap and let a third connection in.""" + host = "192.0.2.7" + w: Any = _PeerWriter() + gate = ListenerAdmission( + transport="TCP", max_connections=10, max_connections_per_host=2, source_ip_allowlist=None + ) + straggler = gate.admit(host) + gate.stopping = True # stop() begins; the straggler's task outlives it + gate.reset() + gate.stopping = False # the same instance restarts + assert gate.check(w, host) is None + current = gate.admit(host) + assert gate.check(w, host) is None + gate.admit(host) + refused = gate.check(w, host) + assert refused is not None and refused.reason == "max_connections_per_host" + assert host in gate.host_capacity_warned + gate.release(host, straggler) + assert gate.per_host == {host: 2}, "a pre-restart release took a slot from the restarted run" + refused = gate.check(w, host) + assert refused is not None and refused.reason == "max_connections_per_host" + assert host in gate.host_capacity_warned, "the warned set belongs to the restarted run too" + # `active` counts sockets, and the straggler's was open until now, whichever run admitted it. + assert gate.active == 2 + # CONTROL: a release from the current run still gives its slot back. + gate.release(host, current) + assert gate.per_host == {host: 1} + assert gate.check(w, host) is None + + +def test_a_release_still_counts_when_a_late_stop_skipped_the_reset() -> None: + """The epoch advances only when reset() really clears the tables. A stop() that finishes after + the restart skips its reset, so the tables still hold the earlier connection's slot, and its + release must give that slot back, or the host would stay one short of its cap for good.""" + host = "192.0.2.7" + gate = ListenerAdmission( + transport="TCP", max_connections=None, max_connections_per_host=2, source_ip_allowlist=None + ) + earlier = gate.admit(host) + gate.stopping = True + gate.stopping = False # restarted before the stop() reached its reset + gate.reset() # the late stop(): a no-op now + gate.admit(host) + assert gate.per_host == {host: 2} + gate.release(host, earlier) + assert gate.per_host == {host: 1} + assert gate.active == 1 + + @pytest.mark.parametrize("kind", ["tcp", "x12", "http"]) async def test_a_connection_reaching_a_stopping_listener_is_refused_unread(kind: str) -> None: events = _Events() From 53c4da33a3e1d5bf38d36e6d654eb416cca98010 Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 20:33:42 -0500 Subject: [PATCH 4/6] FrameCodec refuses 0x20 as a start or end byte (BACKLOG #2900) A reply is neutralised by replacing each frame byte with a space, so a space frame byte would be replaced by itself and the reply could not go as one frame. FrameCodec.__post_init__ now refuses 0x20 for start and end; a 0x20 trailer stays allowed. ADR 0205's reply rule is unchanged. - framing.py: the refusal, the reason once in the class docstring, the frame_neutralised caveat made unconditional, b" " as one constant. - tests: refused directly, via _codec_from_settings and at TcpSource/TcpDestination build, with 0x21 as the control; a neutralised reply decodes as one frame. - CONNECTIONS.md TCP start/end/trailer row, changelog fragment. Proposed PR title: FrameCodec refuses 0x20 as a start or end frame byte (BACKLOG #2900) Proposed banner: CLOSED -- option A, the Manager's decision (batch 265): 0x20 is refused as a start or end byte at codec construction; a 0x20 trailer stays allowed; ADR 0205 unamended. --- changelog.d/2900.changed.md | 3 +++ docs/CONNECTIONS.md | 2 +- messagefoundry/framing.py | 23 ++++++++++++++++++--- tests/test_tcp_transport.py | 41 +++++++++++++++++++++++++++++++++++++ 4 files changed, 65 insertions(+), 4 deletions(-) create mode 100644 changelog.d/2900.changed.md diff --git a/changelog.d/2900.changed.md b/changelog.d/2900.changed.md new file mode 100644 index 0000000000..b932b80375 --- /dev/null +++ b/changelog.d/2900.changed.md @@ -0,0 +1,3 @@ +- **A TCP connection refuses `0x20` (a space) as its `start` or `end` frame byte.** A reply's frame + bytes are replaced by a space, so a space frame byte could not be kept out of a reply. A `0x20` + `trailer` is still allowed. The refusal fires when the connection is built. (`BACKLOG #2900`) diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index 7027092eef..3bc1cbd37a 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -482,7 +482,7 @@ the bytes. | `host` | out | — (required) | the downstream peer to dial. **Inbound takes no host** (wiring error) — listeners bind the service-level `[inbound].bind_host`. | | `port` | both | — (required) | bind/connect port | | `framing` | both | `"stx_etx"` | framing **preset**: `"stx_etx"` (`0x02`/`0x03`, no trailer) or `"vt_fs"`/`"mllp"` (`0x0B`/`0x1C`/`0x0D`). Pass `framing=None` to use explicit bytes instead. | -| `start` / `end` / `trailer` | both | — | explicit delimiter **byte ints** (use with `framing=None`; `trailer` optional). Specifying these *and* a preset is a config error. | +| `start` / `end` / `trailer` | both | — | explicit delimiter **byte ints** (use with `framing=None`; `trailer` optional). Specifying these *and* a preset is a config error. `start` and `end` must differ, and neither may be `0x20` (a space); the connection is refused when it is built. A `0x20` `trailer` is allowed. Why: `FrameCodec` in [`messagefoundry/framing.py`](../messagefoundry/framing.py). | | `encoding` | both | `utf-8` | charset used to encode/decode the framed payload | | `max_connections` | in | `256` | cap on concurrent client connections (flood guard). `None`/`0` = unlimited. | | `receive_timeout` | in | `60.0` | close a client idle this many seconds (slowloris). `None`/`0` = no timeout. | diff --git a/messagefoundry/framing.py b/messagefoundry/framing.py index faaadcca3d..f1cb28abc7 100644 --- a/messagefoundry/framing.py +++ b/messagefoundry/framing.py @@ -91,6 +91,10 @@ class FrameEncodeError(ValueError): :class:`UnicodeEncodeError` quotes the character and carries the whole payload.""" +#: What :meth:`FrameCodec.neutralise` puts in place of a frame byte. +_SPACE = b" " + + def _body(payload: str | bytes, encoding: str) -> bytes: return payload.encode(encoding) if isinstance(payload, str) else bytes(payload) @@ -118,6 +122,12 @@ class FrameCodec: stateful streaming :class:`FrameDecoder`. The ``trailer`` (e.g. MLLP's CR) is **emitted** when framing but **not required** when decoding — a tolerant receiver treats any inter-frame bytes, including a stray trailer after the end delimiter, as noise to discard. + + ``start`` and ``end`` must differ, and neither may be ``0x20``, the space (BACKLOG #2900). + :meth:`neutralise` replaces each frame byte in a reply with a space, so a space frame byte would + be replaced by itself and the reply could not go as one frame; and a delivery holding a space + would be refused under ADR 0205 rule 1. A space trailer is allowed: the start byte is never a + space, so a space trailer cannot open a frame. """ start: int @@ -132,6 +142,12 @@ def __post_init__(self) -> None: raise ValueError(f"frame {name} must be a byte value in 0..255, got {value!r}") if self.start == self.end: raise ValueError("frame start and end delimiters must differ") + for name, value in (("start", self.start), ("end", self.end)): + if value == _SPACE[0]: # the class docstring says why + raise ValueError( + f"frame {name} must not be 0x20 (a space): a space is what replaces a frame " + "byte in a reply, and a space in a message would be taken as a frame byte" + ) def frame(self, payload: str | bytes, encoding: str = "utf-8") -> bytes: """Wrap a message: ``start payload end [trailer]``.""" @@ -167,7 +183,7 @@ def neutralise(self, body: bytes) -> bytes: comes back as is, uncopied; compare with ``!=`` to learn whether anything was replaced.""" if self.start not in body and self.end not in body: return body - return body.replace(bytes([self.start]), b" ").replace(bytes([self.end]), b" ") + return body.replace(bytes([self.start]), _SPACE).replace(bytes([self.end]), _SPACE) def frame_checked( self, payload: str | bytes, encoding: str = "utf-8", *, transport: str = "frame" @@ -184,8 +200,9 @@ def frame_checked( return self.frame(body) def frame_neutralised(self, payload: str | bytes, encoding: str = "utf-8") -> bytes: - """:meth:`frame` after :meth:`neutralise`: one frame for a reply, for any codec whose start - and end bytes are not the space. Raises :class:`FrameEncodeError` as :meth:`frame_checked`.""" + """:meth:`frame` after :meth:`neutralise`, so the framed body holds neither frame byte: no + codec may have the space as its start or end byte (see the class docstring). Raises + :class:`FrameEncodeError` as :meth:`frame_checked`.""" return self.frame(self.neutralise(_encoded(payload, encoding, "reply"))) def decoder(self, max_frame_bytes: int | None = None) -> FrameDecoder: diff --git a/tests/test_tcp_transport.py b/tests/test_tcp_transport.py index 1acb59987a..436eeab1c0 100644 --- a/tests/test_tcp_transport.py +++ b/tests/test_tcp_transport.py @@ -215,6 +215,47 @@ def test_codec_validates_bytes_and_distinct_delimiters() -> None: codec_for(None) # neither preset nor bytes +@pytest.mark.parametrize("role", ["start", "end"]) +def test_codec_refuses_a_space_start_or_end_byte(role: str) -> None: + # BACKLOG #2900. The refusal names the setting, directly, through the settings and at build. + def frame_bytes(byte: int) -> dict[str, int]: + return {"start": 0x02, "end": 0x03, role: byte} + + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + FrameCodec(**frame_bytes(0x20)) + with pytest.raises( + ValueError, match=f"TCP framing misconfigured: frame {role} must not be 0x20" + ): + tcp_mod._codec_from_settings(dict(frame_bytes(0x20))) + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + _source(framing=None, **frame_bytes(0x20)) + with pytest.raises(ValueError, match=f"frame {role} must not be 0x20"): + _dest(9, framing=None, **frame_bytes(0x20)) + # Control: the byte after the space still builds, directly, through the settings and at build. + built = FrameCodec(**frame_bytes(0x21)) + assert getattr(built, role) == 0x21 + assert tcp_mod._codec_from_settings(dict(frame_bytes(0x21))) == built + assert _source(framing=None, **frame_bytes(0x21)).codec == built + assert _dest(9, framing=None, **frame_bytes(0x21)).codec == built + + +def test_codec_allows_a_space_trailer() -> None: + # The trailer is not a frame byte to a decoder, so a space trailer is not refused. + codec = FrameCodec(start=0x02, end=0x03, trailer=0x20) + assert codec.frame_neutralised("A\x02B\x03C") == b"\x02A B C\x03 " + assert tcp_mod._codec_from_settings({"start": 2, "end": 3, "trailer": 0x20}) == codec + + +@pytest.mark.parametrize("trailer", [None, 0x20]) +def test_a_neutralised_reply_decodes_as_one_frame(trailer: int | None) -> None: + # The property behind the 0x20 refusal, for a codec that may be built: a reply holding both + # frame bytes still decodes as exactly one payload with no frame left open. + codec = FrameCodec(start=0x21, end=0x03, trailer=trailer) + decoder = codec.decoder() + assert list(decoder.feed(codec.frame_neutralised("A!B\x03C"))) == [b"A B C"] + assert not decoder.in_frame + + # --- MLLP unchanged after refactor onto the shared codec --------------------- From 155b8602fcb81ba5ec491af0fe54a03565e1c14b Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 20:47:49 -0500 Subject: [PATCH 5/6] Disabled size caps warn at build; File refuses a negative min_age_seconds (BACKLOG #2749 #3151) BACKLOG #2749: HttpSource max_body_bytes, FileSource max_file_bytes and RemoteFileSource max_file_bytes set to 0/None now log a WARNING when the connector is built, naming the connection and the setting, never the value. Same shape as the PR 2292 max_decompressed_bytes warning (not rebuilt). Warn, no refusal, no security_loosenings entry, per the owner's 2026-09-27 ruling on #2006 ("Once the engine at least warns, R5 applies"). Each warning test has a control that the default cap still refuses an over-size input. CONNECTIONS.md cells updated. BACKLOG #3151: FileSource refuses a negative min_age_seconds (value withheld); zero stays the default. The processed_files SQL comment and ADR 0129 section 2 (dated correction) now describe the remote key with the listed mtime (#2758). Timer interval_seconds and the File NaN/inf half were already done by PR 2277. Proposed PR title: Disabled size caps warn at build, File refuses a negative min_age_seconds (BACKLOG #2749 #3151) Banner #2749: CLOSED -- the HTTP max_body_bytes and File/RemoteFile max_file_bytes caps warn at build when turned off; max_decompressed_bytes was #1129's. The 5.2.1 and 2.3.2 re-read is vault work after merge. Banner #3151: CLOSED -- a negative File min_age_seconds is refused at build; the store comment and ADR 0129 describe the remote mtime key. Timer interval_seconds was done by PR 2277. --- changelog.d/2749.security.md | 4 ++ changelog.d/3151.changed.md | 3 + docs/CONNECTIONS.md | 10 +-- ...oss-backend-processed-file-dedup-ledger.md | 8 +++ messagefoundry/store/store.py | 5 +- messagefoundry/transports/file.py | 16 +++++ messagefoundry/transports/http_listener.py | 9 +++ messagefoundry/transports/remotefile.py | 12 +++- tests/test_inbound_http_source.py | 51 ++++++++++++++ tests/test_remotefile_transport.py | 42 ++++++++++++ tests/test_transports.py | 67 +++++++++++++++++-- 11 files changed, 215 insertions(+), 12 deletions(-) create mode 100644 changelog.d/2749.security.md create mode 100644 changelog.d/3151.changed.md diff --git a/changelog.d/2749.security.md b/changelog.d/2749.security.md new file mode 100644 index 0000000000..364d8b3142 --- /dev/null +++ b/changelog.d/2749.security.md @@ -0,0 +1,4 @@ +- **Turning off a size cap on an HTTP, File or SFTP/FTP inbound now logs a warning.** Setting + `max_body_bytes` on an HTTP inbound, or `max_file_bytes` on a File or SFTP/FTP inbound, to `0` or + `None` is still allowed and still removes the cap. The connection now logs a warning each time it + is built, naming the connection and the setting, never the value. (`BACKLOG #2749`) diff --git a/changelog.d/3151.changed.md b/changelog.d/3151.changed.md new file mode 100644 index 0000000000..8daef5eb52 --- /dev/null +++ b/changelog.d/3151.changed.md @@ -0,0 +1,3 @@ +- **A File inbound refuses a negative `min_age_seconds` at build.** A negative value used to build + and act as zero. It is now a build error naming the setting, with the value withheld. Zero, the + default, and any positive value build as before. (`BACKLOG #3151`) diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index 7027092eef..08df5ebab0 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -653,7 +653,7 @@ routes a `Message`; `json`/`xml`/`text`/`fhir` route a `RawMessage` the Handler | `max_connections` | `256` | cap on concurrent clients (connection-flood guard). `None`/`0` = unlimited. | | `receive_timeout` | `60.0` | bound the **whole-request** read — request line + headers + body (slowloris guard); over budget answers a synchronous `408`. `None`/`0` = no timeout. | | `max_connections_per_host` | **off** | cap on concurrent connections from **one peer address**, answered `503` with an `at_capacity` event whose reason reads `max_connections_per_host` (vault BACKLOG #2606). **Ships off, unlike MLLP/TCP/X12**: this connector answers one request per connection, so behind a reverse proxy every partner arrives from one address and any cap would be the listener's whole capacity. Set one only where the listener sees real client addresses. | -| `max_body_bytes` | `16 MiB` | the MLLP frame cap's HTTP twin — an over-declared `Content-Length` is refused `413` **before a body byte is read** (OOM guard). `None`/`0` = unlimited. | +| `max_body_bytes` | `16 MiB` | the MLLP frame cap's HTTP twin — an over-declared `Content-Length` is refused `413` **before a body byte is read** (OOM guard). `None`/`0` = unlimited, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `max_header_bytes` | `64 KiB` | cap the request line + headers (header-flood guard). This one cap can't be switched off: unset or `None` takes the 64 KiB default, and `0` (or `"0"`) is refused at load rather than silently becoming the default (BACKLOG #1872). | | `max_messages_per_second` | **off** | sustained message-rate ceiling for the **whole listener** (ASVS 2.4.1 / 15.2.2, BACKLOG #1114 — the MLLP pacer, ported). Over budget the connector **waits before reading the request**, so the partner is back-pressured and then served in full — **nothing is dropped, refused or answered differently**, and the wait sits outside `receive_timeout` so a paced partner is never handed a `408` for a delay the engine imposed. **Listener-wide, not per-connection**, unlike MLLP/TCP/X12: this connector answers one request per connection, so a per-connection bucket would be charged once and thrown away, bounding nothing. A `GET`/`HEAD` probe waits behind an outstanding debt but **charges nothing**, and neither does a request refused before its body reaches the engine. Only a body the engine reads and records spends budget, including one it then refuses with a `422`, so a peer that submits nothing cannot starve one that does. Unset = no bound, deliberately: a guessed rate throttles real traffic, so the number has to come from your own feed profile. | | `message_burst` | = the rate | tokens the bucket holds, i.e. how large a burst passes unpaced before the sustained rate applies. Only meaningful with `max_messages_per_second` set. Floor of 1 so the listener can always make progress. | @@ -779,11 +779,11 @@ def route(msg): | `directory` | both | — (required) | folder to poll / write into | | `pattern` | in | `*.hl7` | filename glob to pick up | | `poll_seconds` | in | `1.0` | poll interval, a finite number of seconds above zero; anything else is refused at build, including at least zero, a negative value, NaN, infinity and text that is not a number (BACKLOG #2774). It is also the **settle window**: a file is read only once its size and modification time are unchanged since the last poll that saw it (BACKLOG #1811), so every file waits at least one poll. The settle gate is always on and has no setting, but a very small `poll_seconds` narrows its window to almost nothing. | -| `min_age_seconds` | in | `0` | skip files modified within this window. This is an extra wait on top of the settle gate, not the gate itself; set it for a partner that pauses between writes for longer than `poll_seconds`. | +| `min_age_seconds` | in | `0` | skip files modified within this window. This is an extra wait on top of the settle gate, not the gate itself; set it for a partner that pauses between writes for longer than `poll_seconds`. A negative value is refused at build (BACKLOG #3151). | | `after_read` | in | `move` | `move` (→ `.processed`), `delete`, or `leave` (process **in place** — never move/delete the source file, for a read-only share / a directory another system owns; a hashed dedup ledger ensures a left file is ingested **once**, #142) | | `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. | +| `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, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `poll_max_files` | in | `500` | most files one scan will take. The rest stay in the drop directory and the next scan takes them — a **deferral, not a drop**: nothing is quarantined, errored, or left unaccounted for. See [*Per-tick poll ceilings*](#per-tick-poll-ceilings) for the number and when to raise it. `None`/`0` = unlimited. | | `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 | @@ -1055,7 +1055,7 @@ its own policy block below): charged on the handle the read opens, not on an earlier `stat()`: an oversize file is rejected by its handle's size **before** it is read into memory, and the read itself stops at the cap plus one byte, so a file that grows after it was listed is refused too (OOM / DoS guard, BACKLOG #2507). - `None`/`0` disables it. + `None`/`0` disables it, and the connection then logs a `WARNING` each time it is built (BACKLOG #2749). - **Links are refused at listing, read and move time.** The listing looks at each name without following a link there. A drop is read only if the opened handle is a regular file at the listed name inside the watch directory, reached through no symbolic link or junction: POSIX opens each path @@ -1312,7 +1312,7 @@ poll/write shape against a remote server, selected by an internal `protocol` set | `poll_seconds` | in | `5.0` | poll interval, a finite number of seconds above zero; anything else is refused at build, including at least zero, a negative value, NaN, infinity and text that is not a number (BACKLOG #2774). It is also the **settle window**: a file is read only once it lists at the same size, and the same modification time where the server lists one, as at the last poll that saw it (BACKLOG #2071, #2758), so every file waits at least one poll. The gate is always on and has no setting. On a server that lists no modification time (at least FTP without `MLSD`, an `MLSD` server without `modify`, and an SFTP server that omits the times) it reads the listed size alone, so there it cannot see a same-size rewrite, nor anything on a server that lists every file at size 0. | | `min_age_seconds` | in | `0.0` | **accepted but not honoured on a remote source today** — the connector never reads it. Only `File(...)` implements it; the settle gate on `poll_seconds` above, and the partner's own write-then-rename, are what guard against partial reads here. | | `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 + the listed modification time where the server gives one ensures a left file is ingested once; see *Process-in-place* for what a server that lists no modification time misses) | -| `max_file_bytes` | in | `16 MiB` | **charged twice, and the second charge is the one that binds.** Before the retrieve, against the size the **server reported** in its own directory listing — an over-size entry is moved to `error_subdir` without being read. Then **during** the retrieve, against the **bytes actually read**: the download streams in 1 MiB chunks and is cut off at the first byte past the budget, so a share that lists a small file and then delivers an arbitrarily large body is refused mid-transfer rather than buffered whole (BACKLOG #1191). Either refusal quarantines the file to `error_subdir` and logs it — never a silent drop, and never left in place to be re-pulled every poll. `None`/`0` = unlimited, in both charges. | +| `max_file_bytes` | in | `16 MiB` | **charged twice, and the second charge is the one that binds.** Before the retrieve, against the size the **server reported** in its own directory listing — an over-size entry is moved to `error_subdir` without being read. Then **during** the retrieve, against the **bytes actually read**: the download streams in 1 MiB chunks and is cut off at the first byte past the budget, so a share that lists a small file and then delivers an arbitrarily large body is refused mid-transfer rather than buffered whole (BACKLOG #1191). Either refusal quarantines the file to `error_subdir` and logs it — never a silent drop, and never left in place to be re-pulled every poll. `None`/`0` = unlimited, in both charges, and the connection logs a `WARNING` naming the setting each time it is built (BACKLOG #2749). | | `poll_max_files` | in | `500` | most files one poll will take. The rest stay on the share and the next poll takes them — a **deferral, not a drop**. Identical in shape and reasoning to the `File(...)` row; see [*Per-tick poll ceilings*](#per-tick-poll-ceilings). `None`/`0` = unlimited. | | `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`. Instead every send first **lists** `remote_dir`, so the account needs list permission there even with `overwrite = true`, at one extra round trip per delivery. An upload into a vanished dir then fails **retryably** rather than dead-lettering on the partner's permanent no-such-dir; a credential refusal on that listing stops the lane. 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 | diff --git a/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md b/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md index e1f8759844..d211170fc1 100644 --- a/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md +++ b/docs/adr/0129-process-in-place-file-disposition-and-cross-backend-processed-file-dedup-ledger.md @@ -94,6 +94,14 @@ processed_files( opportunistically at the end of a poll tick only when ≥1 new file was recorded, so a stable read-only share (nothing new) never churns the DB. +> **CORRECTION 2026-10-16 (BACKLOG #3151). The remote key above is out of date: since BACKLOG #2758 +> it also folds in the modification time the server lists, where it lists one.** The first bullet's +> reason, that a remote listing "carries no mtime", was wrong for at least SFTP and FTP `MLSD`. A +> listing with no mtime still keys on path + size. The `file_key` comment in the sketch above, +> `sha256(rel-path || mtime || size)`, describes the local key only. `RemoteFileSource._file_key` in +> `messagefoundry/transports/remotefile.py` is the source of record for the remote key, including +> which listings carry no mtime and what each key misses. + ### §3 Store-agnostic transport — a runner-injected ledger seam `transports/` must never import `store/`. The source therefore reaches the ledger through an injected diff --git a/messagefoundry/store/store.py b/messagefoundry/store/store.py index 9f35ada26e..17eb53e728 100644 --- a/messagefoundry/store/store.py +++ b/messagefoundry/store/store.py @@ -6652,8 +6652,9 @@ def _dead_target_pairs(rows: Iterable[tuple[Any, Any]]) -> list[tuple[str, str]] -- files another system owns can be polled without moving/deleting them and a left file is ingested -- ONCE. Shaped like `delivered_keys`/`resend_log`: HASHES + IDS ONLY, no body/PHI, so it is stored in -- the clear (NOT part of the `_cipher` seam). `file_key` is a SHA-256 of the file's identity --- (relative-path+mtime+size local / full-remote-path+size remote — the PATH is folded in, not just the --- basename, so two same-named files under different subdirs stay distinct and are BOTH ingested) — a +-- (relative-path+mtime+size local / full-remote-path+size+the listed mtime remote, the mtime only where +-- the server lists one, BACKLOG #2758 — the PATH is folded in, not just the basename, so two +-- same-named files under different subdirs stay distinct and are BOTH ingested) — a -- DERIVED id, NEVER a cleartext path (a filename/path can embed an MRN) and never logged at INFO+. -- `channel_id` is the inbound connection name (config metadata, non-PHI) — scopes the dedup + the -- age/count prune per connection. The row is INSERTed AFTER the file's message(s) emit successfully (the diff --git a/messagefoundry/transports/file.py b/messagefoundry/transports/file.py index 9a34122e98..a5e4fdde14 100644 --- a/messagefoundry/transports/file.py +++ b/messagefoundry/transports/file.py @@ -677,6 +677,13 @@ def __init__(self, config: Source) -> None: self.min_age_seconds: float = float_setting( s.get("min_age_seconds", 0.0), knob="min_age_seconds", transport="file source" ) + # Zero is the shipped default (no extra wait), so this is not positive_seconds; only a negative + # is refused, as a typo. The scan's `> 0` test would read it as zero (BACKLOG #3151). + if self.min_age_seconds < 0: + raise ValueError( + "file source min_age_seconds must be a number of seconds, zero or more " + "(value withheld)" + ) self.after_read: str = s.get("after_read", "move") # "move" | "delete" | "leave" (#142) if self.after_read not in ("move", "delete", "leave"): raise ValueError( @@ -762,6 +769,15 @@ def __init__(self, config: Source) -> None: # service-account identity, byte-identical. On a non-Windows host a configured credential # raises CredentialUnsupportedError here (a build error), never a silent no-op. self._cred_ctx = _build_credential_context(s) + if self.max_file_bytes is None: + # Off is allowed but never silent (BACKLOG #2749). Logged after every check in this + # constructor, so a source it refuses has not logged it. The line names the setting, + # never its value. + logger.warning( + "file source %s: max_file_bytes is off, so a drop is read whole with no byte " + "ceiling; set a positive value to bound it", + config.name or self.directory, + ) async def _run_fs(self, fn: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: """Run a blocking filesystem callable off the event loop — under the alternate credential (on a diff --git a/messagefoundry/transports/http_listener.py b/messagefoundry/transports/http_listener.py index d767c580e8..4cd084ba1d 100644 --- a/messagefoundry/transports/http_listener.py +++ b/messagefoundry/transports/http_listener.py @@ -775,6 +775,15 @@ def __init__(self, config: Source) -> None: # connections and bound the wait (mirrors MLLPSource, review H-2 / #55). self._clients: set[asyncio.StreamWriter] = set() self._client_tasks: set[asyncio.Task[None]] = set() + if self.max_body_bytes is None: + # Turning the body cap off is documented, so it is allowed, but never silent (BACKLOG + # #2749, as FileSource does for its caps). Logged after every check in this constructor, + # so a source it refuses has not logged it. The line names the setting, never its value. + logger.warning( + "HTTP source %s: max_body_bytes is off, so a request body of any declared size is " + "read whole; set a positive value to bound it", + config.name or f"{self.host}:{self.port}", + ) async def start( self, handler: InboundHandler, *, leader_gate: Callable[[], bool] | None = None diff --git a/messagefoundry/transports/remotefile.py b/messagefoundry/transports/remotefile.py index 30e2a7bdfc..628e1f8c98 100644 --- a/messagefoundry/transports/remotefile.py +++ b/messagefoundry/transports/remotefile.py @@ -47,7 +47,8 @@ the body *before* an ingress row exists, so no admission bound further down the pipeline can see it. The refusal is a content refusal, not a transient one — the file is quarantined to ``error_subdir`` and logged, exactly as an over-listed-size or content-sniff reject is. ``max_file_bytes=0`` disables -both charges (unbounded), which is the operator's explicit choice. +both charges (unbounded), which is the operator's explicit choice, and the source logs a WARNING +saying so each time it is built (BACKLOG #2749). **Idempotency.** Delivery is at-least-once (an upload may re-send) and a poll may re-emit a file that was handled but not yet marked, so downstream consumers **must** tolerate duplicates. @@ -2453,6 +2454,15 @@ def __init__(self, config: Source) -> None: self._other_form_warned_at: float | None = None self._stop = asyncio.Event() self._task: asyncio.Task[None] | None = None + if self._max_file_bytes is None: + # Turning the cap off is documented, so it is allowed, but never silent (BACKLOG #2749, + # as FileSource does for its caps). Logged after every check in this constructor, so a + # source it refuses has not logged it. The line names the setting, never its value. + logger.warning( + "REMOTEFILE source %s: max_file_bytes is off, so a retrieve is read whole with no " + "byte ceiling; set a positive value to bound it", + self._connection or _where(self._host, self._remote_dir), + ) async def start( self, handler: InboundHandler, *, leader_gate: Callable[[], bool] | None = None diff --git a/tests/test_inbound_http_source.py b/tests/test_inbound_http_source.py index 7eba06389c..1e99f0c77e 100644 --- a/tests/test_inbound_http_source.py +++ b/tests/test_inbound_http_source.py @@ -505,6 +505,57 @@ async def test_read_request_rejects_declared_oversize() -> None: assert exc.value.status == 413 and exc.value.kind == "frame_oversize" +def _http_source(**over: Any) -> HttpSource: + ic = build_inbound_connection("IB_HTTP", Http(port=0, **over), router="r") + return HttpSource(_source_config(ic, "127.0.0.1", {})) + + +def _body_cap_off_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + r + for r in caplog.records + if r.levelno == logging.WARNING and "max_body_bytes is off" in r.getMessage() + ] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_http_source_warns_once_when_max_body_bytes_is_off( + off: object, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #2749: off is documented and allowed, but never silent. The line names the + connection and the setting, and withholds the value. + + Red mutation: drop the warning from ``HttpSource.__init__``. No record is found.""" + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.http_listener"): + src = _http_source(max_body_bytes=off) + assert src.max_body_bytes is None + [record] = _body_cap_off_warnings(caplog) + assert "IB_HTTP" in record.getMessage() + + +async def test_http_source_default_max_body_bytes_warns_nothing_and_still_refuses_oversize( + store: MessageStore, caplog: pytest.LogCaptureFixture +) -> None: + """The control for the warning above, through the running listener: the shipped default says + nothing, and a body declared one byte past it is refused ``413`` before a body byte is sent.""" + ic = build_inbound_connection( + "IB_HTTP", Http(port=0), router="r", content_type=ContentType.TEXT + ) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.http_listener"): + src = await _start_source(store, ic) + try: + assert _body_cap_off_warnings(caplog) == [] + declared = DEFAULT_MAX_BODY_BYTES + 1 + head = f"POST /ingest HTTP/1.1\r\nHost: localhost\r\nContent-Length: {declared}\r\n\r\n" + resp = await _http(src.sockport, raw_override=head.encode("ascii")) + finally: + await src.stop() + assert resp.status == 413 + cur = await store._db.execute("SELECT COUNT(*) AS n FROM messages") + row = await cur.fetchone() + assert row is not None and row["n"] == 0 # refused before any ingress row + + async def test_read_request_rejects_chunked() -> None: raw = b"POST /x HTTP/1.1\r\nHost: h\r\nTransfer-Encoding: chunked\r\n\r\n" with pytest.raises(HttpRequestError) as exc: diff --git a/tests/test_remotefile_transport.py b/tests/test_remotefile_transport.py index 6f87965778..df3e30ab39 100644 --- a/tests/test_remotefile_transport.py +++ b/tests/test_remotefile_transport.py @@ -853,6 +853,48 @@ async def test_source_zero_max_file_bytes_keeps_the_retrieve_unbounded( assert len(h.bodies) == 1 # delivered whole — the operator disabled the cap +def _file_cap_off_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + r + for r in caplog.records + if r.levelno == logging.WARNING and "max_file_bytes is off" in r.getMessage() + ] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_source_warns_once_when_max_file_bytes_is_off( + off: object, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """BACKLOG #2749: off is documented and allowed, but never silent. The line names where the + source polls and the setting, and withholds the value. + + Red mutation: drop the warning from ``RemoteFileSource.__init__``. No record is found.""" + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + src = _src(monkeypatch, _FakeClient(), max_file_bytes=off) + assert src._max_file_bytes is None + [record] = _file_cap_off_warnings(caplog) + assert "sftp.example.com:/in" in record.getMessage() + + +async def test_source_default_max_file_bytes_warns_nothing_and_still_refuses_oversize( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """The control for the warning above: the shipped default says nothing, and it still refuses a + file listed one byte past it, without retrieving it.""" + name = "/in/big.hl7" + client = _FakeClient(files={name: b"x"}, sizes={name: DEFAULT_MAX_FILE_BYTES + 1}) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.remotefile"): + src = _src(monkeypatch, client) + assert _file_cap_off_warnings(caplog) == [] + h = _RecordingHandler() + src._handler = h + await _settle(src) + await src._poll_once() + assert h.bodies == [] + assert not any(op == "retrieve" for op, _ in client.ops) + assert "/in/.error/big.hl7" in client.files + + def test_sftp_retrieve_stops_reading_past_the_budget(monkeypatch: pytest.MonkeyPatch) -> None: """The SFTP backend reads in chunks and refuses mid-transfer, so a hostile body is never buffered whole. ``read_total`` proves the bytes were never pulled.""" diff --git a/tests/test_transports.py b/tests/test_transports.py index 41ecba95ae..6a13cb3d24 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -1749,11 +1749,11 @@ def test_file_source_member_cap_refuses_zero_and_bad_values(bad: object) -> None FileSource(Source(type=ConnectorType.FILE, settings=settings)) -def _decompressed_cap_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: +def _cap_off_warnings(caplog: pytest.LogCaptureFixture, knob: str) -> list[logging.LogRecord]: return [ r for r in caplog.records - if r.levelno == logging.WARNING and "max_decompressed_bytes is off" in r.getMessage() + if r.levelno == logging.WARNING and f"{knob} is off" in r.getMessage() ] @@ -1766,7 +1766,7 @@ def test_file_source_warns_once_when_the_decompressed_byte_cap_is_off( with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): src = FileSource(Source(type=ConnectorType.FILE, name="IB_DEMO_ADT", settings=settings)) assert src.max_decompressed_bytes is None - [record] = _decompressed_cap_warnings(caplog) + [record] = _cap_off_warnings(caplog, "max_decompressed_bytes") assert "IB_DEMO_ADT" in record.getMessage() @@ -1784,7 +1784,66 @@ def test_file_source_does_not_warn_while_the_decompressed_byte_cap_holds( # Without decompress the cap is never read, so turning it off changes nothing to warn about. with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): FileSource(Source(type=ConnectorType.FILE, settings={"directory": "x", **settings})) - assert _decompressed_cap_warnings(caplog) == [] + assert _cap_off_warnings(caplog, "max_decompressed_bytes") == [] + + +@pytest.mark.parametrize("off", [0, None, "0"], ids=["zero", "none", "zero-text"]) +def test_file_source_warns_once_when_max_file_bytes_is_off( + off: object, caplog: pytest.LogCaptureFixture +) -> None: # #2749 + # Off is documented and allowed, but never silent. The line names the connection and the setting. + # Red mutation: drop the warning from FileSource.__init__. No record is found. + settings = {"directory": "x", "max_file_bytes": off} + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + src = FileSource(Source(type=ConnectorType.FILE, name="IB_DEMO_ADT", settings=settings)) + assert src.max_file_bytes is None + [record] = _cap_off_warnings(caplog, "max_file_bytes") + assert "IB_DEMO_ADT" in record.getMessage() + + +async def test_file_source_default_max_file_bytes_warns_nothing_and_still_refuses_oversize( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: # #2749 + # The control for the warning above: the shipped default says nothing, and a drop one byte past it + # is still quarantined unread. + inbox = tmp_path / "in" + inbox.mkdir() + (inbox / ".error").mkdir() # start() would make it; _scan_once is driven directly here + with (inbox / "big.hl7").open("wb") as fh: + fh.truncate(DEFAULT_MAX_FILE_BYTES + 1) + with caplog.at_level(logging.WARNING, logger="messagefoundry.transports.file"): + src = _file_source_for(inbox) + assert _cap_off_warnings(caplog, "max_file_bytes") == [] + received: list[bytes] = [] + + async def handler(raw: bytes) -> None: + received.append(raw) + + src._handler = handler + await src._scan_once() # the settle poll (BACKLOG #1811): records the stat, reads nothing + await src._scan_once() + assert received == [] + assert (inbox / ".error" / "big.hl7").exists() + + +@pytest.mark.parametrize("bad", [-1, -0.5, "-1", "-0.5"]) +def test_file_source_refuses_a_negative_min_age_seconds(bad: object) -> None: # #3151 + # Red mutation: drop the negative check from FileSource.__init__. Every case builds. + settings = {"directory": "x", "min_age_seconds": bad} + refusal = ( + "file source min_age_seconds must be a number of seconds, zero or more (value withheld)" + ) + with pytest.raises(ValueError, match=f"^{re.escape(refusal)}$"): + FileSource(Source(type=ConnectorType.FILE, settings=settings)) + + +@pytest.mark.parametrize("ok", ["unset", 0, 0.0, "0", "-0", 2.5, "3600"]) +def test_file_source_accepts_zero_or_a_positive_min_age_seconds(ok: object) -> None: # #3151 + # The control for the refusal above: unset (the shipped default of zero), a zero in any exact + # spelling, a negative zero, and a positive value all build. + settings = {"directory": "x"} if ok == "unset" else {"directory": "x", "min_age_seconds": ok} + src = FileSource(Source(type=ConnectorType.FILE, settings=settings)) + assert src.min_age_seconds == (0.0 if ok == "unset" else float(str(ok))) async def test_file_source_quarantines_corrupt_archive(tmp_path: Path) -> None: From 389c4a65da0ead8185e511ea825c0b3a7a145565 Mon Sep 17 00:00:00 2001 From: Scott Hall Date: Fri, 16 Oct 2026 21:17:38 -0500 Subject: [PATCH 6/6] Code-review fixes for BACKLOG #2938 and BACKLOG #2939: docs, an empty read, an end-to-end straggler test - docs/CONNECTIONS.md: the max_frame_seconds bullet that called the X12 newline-after-IEA a known gap now states the two-byte allowance (#2938). - X12FrameReader.feed(b"") leaves the line-end allowance alone, as FrameDecoder does; the reader test pins it. The comment on the local tail constants now gives the real reason they are copies: parsing may not import messagefoundry.framing (tests/test_dependency_boundaries.py). feed() says the previous call's generator must be drained first. - ListenerAdmission.release states, once, the limit the epoch leaves: while a straggler is still open its host may hold one socket over max_connections_per_host, since reset() already dropped its slot. The MLLP stop() comment links to it rather than restating it. - New test, mllp/tcp/x12: a client held in its handler outlives a stop() cut short, the same instance restarts and re-admits the host, then the straggler releases; the restarted count still holds its connection. Each listener fails it when its release passes the current epoch instead of its own. Proposed PR title: X12 trailer newline no longer starts the frame deadline; a post-restart straggler no longer undercounts the per-host cap (BACKLOG #2938 #2939) Proposed banner, BACKLOG #2938: SHIPPED -- X12FrameReader exposes trailer_only, set eagerly in feed(): true only for a read of at most two CR/LF bytes directly after a completed interchange's full terminator; any other byte ends the allowance. The X12 listener passes it to FrameClock.after_read. x12 rows with controls (a non-line-end byte, three line ends) in the shared line-end test; a reader unit test; docs/CONNECTIONS.md updated. framing.py untouched. Left open: the outbound reuse path's leftover flag still reads a trailing CR/LF as a desync (out of scope by Manager decision). Proposed banner, BACKLOG #2939: SHIPPED -- ListenerAdmission.admit returns an epoch, release requires it, reset advances it past its stopping guard. A release from an earlier epoch lowers `active` only (closing step 3: `active` counts open sockets across runs, so it still drops). All four listeners pass the token; MLLP, raw TCP and X12 are pinned end to end, HTTP by mypy only. Left open: a live straggler is not counted per host after the reset, so its host may hold one socket over the cap until it closes. --- docs/CONNECTIONS.md | 6 +-- messagefoundry/parsing/x12/interchange.py | 20 +++++++--- messagefoundry/transports/admission.py | 4 +- messagefoundry/transports/mllp.py | 3 +- tests/test_listener_admission.py | 48 +++++++++++++++++++++++ tests/test_x12_parsing.py | 6 +++ 6 files changed, 77 insertions(+), 10 deletions(-) diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index 7027092eef..0088a151ad 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -3581,9 +3581,9 @@ Facts that are easy to get wrong, stated plainly first: Noise in the same read as a completed frame starts nothing. A sender that holds a quiet socket open with them is closed at `max_frame_seconds`, even where `receive_timeout` is longer or off. - * On X12 the frame is the interchange, from its `ISA` to its `IEA`. X12 has no line-end allowance - yet, so a newline after the `IEA` that arrives in a read of its own starts the clock. That is a - known gap, not the intended rule. + * On X12 the frame is the interchange, from its `ISA` to the terminator of its `IEA` segment. Up + to two bytes directly after that terminator belong to the interchange when each is CR or LF, so + a newline after the `IEA` starts nothing, however late it arrives (vault BACKLOG #2938). * Time the engine withholds, for pacing or an intake pause, is not counted. Vault BACKLOG #2847 moved MLLP onto this rule. Its clock used to start only at a start byte. diff --git a/messagefoundry/parsing/x12/interchange.py b/messagefoundry/parsing/x12/interchange.py index cb38dc5050..95b1d0c518 100644 --- a/messagefoundry/parsing/x12/interchange.py +++ b/messagefoundry/parsing/x12/interchange.py @@ -47,8 +47,11 @@ #: ``IEA*...~``. Up to :data:`_FRAME_TAIL_MAX` of them directly after a completed interchange are its #: ending, however reads split them (vault BACKLOG #2938). The cap keeps a peer from trickling them #: to hold a listener's frame deadline off: past it they are inter-interchange noise like any byte. -#: The same rule as ``messagefoundry.framing.FrameDecoder``'s, kept local so this pure parsing module -#: imports nothing from the transport layer. +#: The same rule as ``messagefoundry.framing.FrameDecoder``'s, written as a local copy because +#: ``parsing`` may not import ``messagefoundry.framing`` (its import allowlist in +#: tests/test_dependency_boundaries.py). So the two can drift; the x12 and MLLP rows of +#: ``test_line_end_bytes_ending_a_frame_do_not_start_the_frame_deadline`` in +#: tests/test_listener_admission.py pin that both allow two line ends and cut on a third. _FRAME_TAIL_BYTES = b"\r\n" _FRAME_TAIL_MAX = 2 @@ -130,11 +133,18 @@ def feed(self, data: bytes) -> Iterator[bytes]: """Yield each interchange completed by ``data``, carrying any partial one to the next call. :attr:`trailer_only` is decided here, eagerly, so it never describes an earlier read; the - reassembly itself still runs lazily as the generator is iterated. + reassembly itself still runs lazily as the generator is iterated. The allowance it reads is + set as an interchange is taken, so drain one call's generator before the next call, as the + listener does; a caller that stops early reads the next call's flag against a stale + allowance. """ self.trailer_only = bool(data) and _only_line_ends(data, self._tail_left) - # Any other byte ends the allowance: what follows it is not the interchange's ending. - self._tail_left = self._tail_left - len(data) if self.trailer_only else 0 + # Any other byte ends the allowance: what follows it is not the interchange's ending. An + # empty read holds no byte, so it leaves the allowance as it was, as FrameDecoder does. + if data and not self.trailer_only: + self._tail_left = 0 + else: + self._tail_left -= len(data) return self._frames(data) def _frames(self, data: bytes) -> Iterator[bytes]: diff --git a/messagefoundry/transports/admission.py b/messagefoundry/transports/admission.py index 453648bb40..7bc75aed07 100644 --- a/messagefoundry/transports/admission.py +++ b/messagefoundry/transports/admission.py @@ -195,7 +195,9 @@ def release(self, peer_host: str | None, epoch: int) -> None: already dropped its per-host slot, and the table now counts the restarted listener's connections, so it touches neither that table nor the warned set (vault BACKLOG #2939). It still gives back ``active``, which :meth:`reset` leaves alone: that socket was open, and - counted, until now, whichever run admitted it.""" + counted, until now, whichever run admitted it. The per-host cap is not exact across a + restart: while such a straggler is still open, its host may hold one socket over + ``max_connections_per_host`` for it, since the reset already dropped its slot.""" self.active -= 1 if peer_host is None or epoch != self._epoch: return diff --git a/messagefoundry/transports/mllp.py b/messagefoundry/transports/mllp.py index 6cd5c7f0ce..f1f4471b3a 100644 --- a/messagefoundry/transports/mllp.py +++ b/messagefoundry/transports/mllp.py @@ -2196,7 +2196,8 @@ async def stop(self) -> None: # epoch of the run that admitted it, and `reset` advances the epoch, so that release # leaves the restarted listener's per-host count alone rather than undercounting it # (vault BACKLOG #2939). `active` still drops on it: that socket was open until then, - # and the global cap that predates this row reads it. + # and the global cap that predates this row reads it. What the epoch leaves open while + # a straggler is still live is stated once, in `ListenerAdmission.release`. self._admission.reset() # Now that no client handlers are in flight, this should complete promptly — but on the Windows # ProactorEventLoop a still-pending overlapped accept/read can make wait_closed() never return, diff --git a/tests/test_listener_admission.py b/tests/test_listener_admission.py index cd262599cd..93c477ddaa 100644 --- a/tests/test_listener_admission.py +++ b/tests/test_listener_admission.py @@ -765,6 +765,54 @@ async def failing_wait(fs: Any, *args: Any, **kwargs: Any) -> Any: await asyncio.wait_for(source.stop(), timeout=10.0) +@pytest.mark.parametrize("kind", ["mllp", "tcp", "x12"]) +async def test_a_straggler_releasing_after_a_restart_leaves_the_new_per_host_count( + kind: str, +) -> None: + """vault BACKLOG #2939, end to end through the listener's own admit and release. A client held + in its handler outlives a stop() cut short; the same instance restarts and admits the host + again; then the straggler releases. The restarted run's count must still hold its connection. + This pins the MLLP, raw-TCP and X12 listeners passing on the epoch their admit returned. The + HTTP listener's handler has another shape and is not driven here.""" + entered = asyncio.Event() + release = asyncio.Event() + + async def handler(raw: bytes) -> str: + entered.set() + await release.wait() + return build_ack(raw, code="AA") + + source = _build(kind) + await source.start(handler) + gate = source._admission # type: ignore[attr-defined] + writers: list[asyncio.StreamWriter] = [] + try: + _reader, old = await asyncio.open_connection("127.0.0.1", source.sockport) # type: ignore[attr-defined] + writers.append(old) + old.write(_whole_frame(kind, HL7)) + await old.drain() + await asyncio.wait_for(entered.wait(), 3.0) + stopping = asyncio.ensure_future(source.stop()) + await asyncio.sleep(0.1) # inside the client-task grace wait + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + assert gate.per_host == {} # the reset ran; the straggler is still in its handler + await source.start(handler) # the same instance restarts + _reader, new = await asyncio.open_connection("127.0.0.1", source.sockport) # type: ignore[attr-defined] + writers.append(new) + assert await _until(lambda: gate.per_host == {"127.0.0.1": 1}) + assert gate.active == 2 + release.set() + assert await _until(lambda: gate.active == 1), "the straggler never released" + assert gate.per_host == {"127.0.0.1": 1}, "the straggler took the restarted run's slot" + finally: + release.set() + for writer in writers: + await _close(writer) + await asyncio.wait_for(source.stop(), timeout=10.0) + + # --- A normal frame still flows, and is answered, with every new bound at its default ------------- diff --git a/tests/test_x12_parsing.py b/tests/test_x12_parsing.py index bd581d3ffd..c83cace92c 100644 --- a/tests/test_x12_parsing.py +++ b/tests/test_x12_parsing.py @@ -702,6 +702,12 @@ def test_frame_reader_marks_only_an_interchanges_own_line_ends_as_trailer_only() assert list(reader.feed(data)) == [data] assert list(reader.feed(b"\r\n")) == [] assert reader.trailer_only + # An empty read is not trailer-only, and holds no byte to end the allowance with. + assert list(reader.feed(data)) == [data] + assert list(reader.feed(b"")) == [] + assert not reader.trailer_only + assert list(reader.feed(b"\r")) == [] + assert reader.trailer_only # CONTROL: three in one read are past the allowance. assert list(reader.feed(data)) == [data] assert list(reader.feed(b"\r\n\r")) == []