conformance: read the archive, and the loss it admits to

README.md

@@ -29,7 +29,7 @@ See [VERSIONING.md](./VERSIONING.md) for the change classification, the compatib
 
 | Tool | Description |
 |------|-------------|
-| [dz-conformance](./tools/conformance/) | Conformance subscriber for publishers: validates one feed — live multicast or pcap replay — against an explicit rule catalog drawn from these specs, and returns a CI-friendly exit code |
+| [dz-conformance](./tools/conformance/) | Conformance subscriber for publishers: validates one feed — live multicast or capture replay (`pcap` or `pcapng`) — against an explicit rule catalog drawn from these specs, and returns a CI-friendly exit code |
 
 `dz-conformance` covers the Top-of-Book, Midpoint, Market-by-Order and Market-by-Price feeds, together with the Reference Data supplement they share. The Order-Intent and Perp Stats feeds have no checker yet. Per-feed rule counts and the known coverage gaps are in its [README](./tools/conformance/README.md).
 

tools/conformance/core/finding.go

@@ -86,6 +86,13 @@ const (
 	// ReasonLoss — a datagram went missing and could have carried what the check
 	// needed. Never a publisher fault.
 	ReasonLoss = "loss"
+	// ReasonCaptureLoss — the *capture* admits it failed to record datagrams in
+	// this window (pcapng's epb_dropcount), so what the check needed may never
+	// have reached the file. Distinct from ReasonLoss because the owner is
+	// different: the recorder, not the wire. An offline replay that charges its
+	// own recorder's drops to the publisher is the single error keeping the bytes
+	// was meant to prevent, so the two causes are not allowed to share a label.
+	ReasonCaptureLoss = "capture_loss"
 	// ReasonColdStart — the subscriber joined mid-stream, or the state the check
 	// reads has not been established yet.
 	ReasonColdStart = "cold_start"
@@ -121,7 +128,7 @@ const (
 
 // reasons is the closed set above, for validation.
 var reasons = map[string]struct{}{
-	ReasonLoss: {}, ReasonColdStart: {}, ReasonReorder: {}, ReasonPending: {},
+	ReasonLoss: {}, ReasonCaptureLoss: {}, ReasonColdStart: {}, ReasonReorder: {}, ReasonPending: {},
 	ReasonOverflow: {}, ReasonTruncated: {}, ReasonInsufficientWindow: {},
 	ReasonSuperseded: {}, ReasonUntrusted: {}, ReasonBoundSubset: {},
 	ReasonTransition: {}, ReasonUnspecified: {},

tools/conformance/engine/engine.go

@@ -47,6 +47,18 @@ type Engine struct {
 	// The refdata state machine reads it to tell the shutdown ManifestSummary the
 	// spec mandates from a mid-service Valid drop; see resolveValidZero.
 	sessionEnd map[uint8]struct{}
+	// captureLossEpoch counts the admissions ObserveCaptureLoss has taken. A
+	// series that carries a stale value has had a drop admitted since its last
+	// message and so has one gap the drop can account for; one that carries the
+	// current value has already been given it. It is a counter rather than a flag
+	// because the excuse is per series and the series are not enumerable when the
+	// admission arrives — see ObserveCaptureLoss.
+	captureLossEpoch uint64
+	// curCaptureEpoch is the frame-being-classified's stamped epoch, the same
+	// per-frame context curUnknownSchema is. The per-instrument rules read it
+	// rather than captureLossEpoch: they run inside classify, where the engine's
+	// own counter may already have moved on for frames still in the buffer.
+	curCaptureEpoch uint64
 }
 
 // New constructs an Engine with the given config and reporter.
@@ -130,6 +142,17 @@ func (e *Engine) Emit(ruleID string, st core.Status, port core.Port, seq uint64,
 	if len(reason) > 0 {
 		rsn = reason[0]
 	}
+	// Name the owner of the loss. Every gated rule reports ReasonLoss for "a
+	// datagram that could have carried what I needed is missing", and on a replay
+	// that datagram may be one the recorder admits it never wrote (see
+	// ObserveCaptureLoss). The distinction is not cosmetic: `loss` sends an
+	// operator to the network and `capture_loss` sends them to the recorder, and
+	// the rules cannot each make the call — the taint is on the window, not on the
+	// individual gap. Substituting here, at the one point every finding passes
+	// through, keeps that knowledge in one place.
+	if st == core.Unverifiable && rsn == core.ReasonLoss && e.captureDirtyOn(port, ch) {
+		rsn = core.ReasonCaptureLoss
+	}
 	e.rep.Record(core.Finding{RuleID: ruleID, Severity: sev, Status: st, Feed: e.cfg.Feed,
 		Port: port, Seq: seq, ChannelID: ch, InstrumentID: inst, Detail: detail, Reason: rsn, At: e.now()})
 }
@@ -270,6 +293,77 @@ func (e *Engine) taintPortWide(port core.Port) {
 	}
 }
 
+// ObserveCaptureLoss records datagrams the *capture* admits it failed to record
+// before the datagram about to be processed — pcapng's epb_dropcount, which is
+// the only place an archive says what it is missing.
+//
+// **Why it taints everything.** The recorder drops at its interface: what it
+// lost was never parsed, so its channel, its source address and even its
+// destination port are unknowable. Marking only the instance whose datagram
+// carried the admission would leave every other series reading a
+// capture-inflicted gap as the publisher's, which is precisely the error the
+// archive format was chosen to prevent — one lost datagram breaks the
+// per-instrument sequence chain of every instrument it carried, so the
+// misattribution amplifies rather than staying proportional. The direction is
+// the safe one taintPortWide already takes: the cost of tainting a series the
+// drop did not touch is a rule that grades Unverifiable instead of pass, never a
+// Violation the publisher did not commit.
+//
+// An instance first seen *after* the drop is deliberately not tainted. It seeds
+// its own sequence baseline from its first datagram, so loss that precedes it
+// declares no gap and is already reported as a mid-stream join (cold_start).
+//
+// **What it does not do is last.** A drop can account for the first gap each
+// series shows after it and for nothing beyond that, so the two things it sets
+// have bounded lifetimes rather than running to the end of the era:
+// captureDirty, which names the owner of a gap, is spent on the next frame
+// classified on its series (classify), and captureLossEpoch, which is what the
+// per-instrument density rule consults, is spent per instrument on that
+// instrument's next observed step (checkMBPSeq). Without those bounds one
+// admitted drop silences a MUST rule for the rest of the segment and relabels
+// every later network gap as the recorder's, which sends an operator to the
+// wrong machine.
+func (e *Engine) ObserveCaptureLoss(n uint64) {
+	if n == 0 {
+		return
+	}
+	e.captureLossEpoch++
+	for _, pt := range e.ports {
+		pt.dirtyWindow = true
+		pt.captureDirty = true
+	}
+	// A snapshot group in flight is not touched here, and cannot be: the reorder
+	// buffer means the group may not have been opened yet when its own loss is
+	// admitted. It reads the counter above against the epoch it opened at
+	// instead — see captureLossSince.
+}
+
+// captureLossSince reports whether the capture has admitted a drop since a
+// window opened at the given epoch.
+//
+// The counterpart of the per-frame comparison in classify, for a window that is
+// not a frame: an unclosed snapshot group at end of run has no next frame to be
+// judged against, and the drop that cost it its SnapshotEnd may have been
+// admitted after the last datagram the run yielded. Comparing epochs at the
+// point of judgement rather than tainting the group when the drop arrives is
+// what makes it work through the reorder buffer, which can open the group after
+// the admission that concerns it.
+func (e *Engine) captureLossSince(epoch uint64) bool {
+	return e.captureLossEpoch != epoch
+}
+
+// captureDirtyOn reports whether any instance of this (port, channel) has a
+// window that admitted capture loss. It ORs across the channel's instances for
+// the same reason dirtyOn does, and is read only to label a finding's cause.
+func (e *Engine) captureDirtyOn(port core.Port, ch uint8) bool {
+	for k, pt := range e.ports {
+		if k.port == port && k.ch == ch && pt.captureDirty {
+			return true
+		}
+	}
+	return false
+}
+
 // mktdataPending reports whether any mktdata reorder buffer on this channel still
 // holds unclassified frames. Cross-port snapshot checks that compare a snapshot against
 // mktdata-derived state (SNAP.ANCHOR_IS_MKTDATA_SEQ, SNAP.LAST_INSTRUMENT_SEQ_
@@ -300,7 +394,7 @@ func (e *Engine) snapPortDirty(ch uint8) bool {
 // buffer in seq order.
 func (e *Engine) Process(src netip.Addr, f *wire.Frame, port core.Port, sf []wire.StructFinding) {
 	pt := e.instanceTrack(src, port, f.Header.ChannelID)
-	tuple := intakeTuple{frame: f, port: port, structFindings: sf}
+	tuple := intakeTuple{frame: f, port: port, structFindings: sf, captureEpoch: e.captureLossEpoch}
 
 	res := pt.enqueue(tuple, e.cfg.ReorderWindow)
 	if res.quarantine {
@@ -404,6 +498,20 @@ func (e *Engine) classify(item *bufferItem, pt *portTracker) {
 	// and new-era items are classified after. So seq tracking is always active.
 	e.beginFrame(f.Header.SchemaVersion)
 
+	// Whether an admitted capture drop can own what this frame shows: one
+	// arrived between the previous frame on this series and this one. Recomputed
+	// per frame rather than left standing from ObserveCaptureLoss, because a drop
+	// explains the first gap after it and nothing later — a taint that ran to the
+	// end of the era renamed every subsequent network gap `capture_loss` and sent
+	// an operator to the recorder for loss the wire caused.
+	//
+	// Read off the frame's own stamped epoch, not the engine's current one: the
+	// reorder buffer means frames are classified after later ones have arrived,
+	// and an admission that landed while this frame was buffered belongs to the
+	// frame behind it, not to this one.
+	pt.captureDirty = item.tuple.captureEpoch > pt.captureEpochSeen
+	e.curCaptureEpoch = item.tuple.captureEpoch
+
 	// Channel-wide reset bookkeeping for MBP, driven by every accepted frame on any
 	// port rather than by the per-series era advance in Process.
 	e.mbpObserveEra(port, f.Header.ChannelID, item.era)
@@ -502,6 +610,20 @@ func (e *Engine) classify(item *bufferItem, pt *portTracker) {
 	if port == core.PortSnapshot && e.cfg.Feed == core.FeedMBP {
 		e.checkMBPSnapshot(f, f.Header.ChannelID, f.Header.Sequence)
 	}
+
+	// The admissions up to this frame's arrival are now spent on this series: the
+	// next frame's own stamp is what decides whether a drop can own what *it*
+	// shows. Only a frame that advanced the series counts — a backward or
+	// duplicate frame shows nothing about what follows a gap. Last in the
+	// function, after everything that reads the flag above.
+	//
+	// dirtyWindow is deliberately not spent with it: that flag answers "could a
+	// gap explain this?", which stays true for the rest of the era once a gap has
+	// been seen (see taintOn), while this one answers the narrower "was the
+	// recorder's admitted drop what that gap was?".
+	if pt.lastSeq != nil && *pt.lastSeq == f.Header.Sequence {
+		pt.captureEpochSeen = item.tuple.captureEpoch
+	}
 }
 
 // observeSessionEnd records that a channel announced the end of its session.

tools/conformance/engine/gate.go

@@ -167,6 +167,11 @@ type openSnapshot struct {
 	// dirty is true if a snapshot-port seq gap was observed DURING this group
 	// (i.e. after this group's SnapshotBegin). Pre-group gaps do not set dirty.
 	dirty bool
+	// captureEpoch is the arrival-time capture-loss epoch this group opened at.
+	// A drop admitted after it is loss inside this group's window — and one that
+	// cost the group its SnapshotEnd leaves no snapshot-port gap for dirty to
+	// see, because the group simply never closes. See Engine.captureLossSince.
+	captureEpoch uint64
 	// structuralViolation is true if any structural violation was detected during
 	// this group (ORDER_SNAPSHOT_ID_MATCH, SNAPSHOT_ORDER_NO_DUP_ORDER_ID,
 	// EMPTY_BOOK_WELL_FORMED, etc.). The oracle gates on this flag to avoid
@@ -721,14 +726,19 @@ func (e *Engine) applyDeltaSeq(ch uint8, instrID uint32, perSeq uint32, frameSeq
 		// without a reset count change) is handled in the default arm. But a
 		// forward jump alone is not a snapshot-reset issue.
 
-		st := core.Violation
+		st, reason := core.Violation, ""
 		if !gapless {
-			// A mktdata channel gap could account for the missing delta(s).
-			st = core.Unverifiable
+			// A mktdata channel gap could account for the missing delta(s). The cause
+			// is named rather than left empty because it is the label an operator
+			// reads off `unverifiable_total{reason}` — and because Emit re-owns it as
+			// `capture_loss` when the capture admits the datagram never reached the
+			// file, which is the distinction that keeps a replay from charging its own
+			// recorder's drops to the publisher.
+			st, reason = core.Unverifiable, core.ReasonLoss
 		}
 		e.Emit("DELTA.PERINSTR_DENSITY", st, core.PortMktData, frameSeq, ch, instrID,
 			fmt.Sprintf("instrument %d: per-instrument seq jumped %d→%d (expected %d)",
-				instrID, last, perSeq, last+1))
+				instrID, last, perSeq, last+1), reason)
 
 	default:
 		// perSeq <= last: duplicate or late arrival.
@@ -896,6 +906,7 @@ func (e *Engine) handleSnapBegin(m wire.Message, ch uint8, snapPortSeq uint64) {
 		orderIDs:        make(map[uint64]struct{}),
 		orders:          make(map[uint64]snapOrderRecord),
 		dirty:           false,
+		captureEpoch:    e.curCaptureEpoch,
 		lastSnapPortSeq: snapPortSeq,
 	}
 }
@@ -1206,7 +1217,13 @@ func (e *Engine) flushOpenSnaps() {
 			continue
 		}
 		st := core.Violation
-		if open.dirty {
+		// The group's own flag catches a snapshot-port gap during its lifetime.
+		// It cannot catch the datagram the capture admits it never wrote: that
+		// leaves no gap, the group just never closes, and at end of run no later
+		// frame can arrive to taint anything. Loss admitted after the last mapped
+		// datagram is inside this window too, which is why the run loop hands the
+		// residual over before this runs.
+		if open.dirty || e.captureLossSince(open.captureEpoch) {
 			st = core.Unverifiable
 		}
 		e.Emit("SNAP.BEGIN_ORDER_END_GROUPING", st, core.PortSnapshot, 0, ch, open.instrID,