From d81186b64c1e0f807cf17726b684d117655a9eea Mon Sep 17 00:00:00 2001 From: Andrew McConnell Date: Mon, 10 Aug 2026 18:28:37 -0500 Subject: [PATCH 1/2] marketbyorder-bot: track Reset Count per channel Port of the market-by-price bot fix. Reset Count is per publisher, and a group can carry two publishers interleaved on the same ports, so two steady-but-different values must not be read as a reset. Key resetCount by channel_id and scope the wipe to the resetting channel. This shard also owns snapCtx, keyed by (channel, snapshot_id), so resetChannel clears that too. --- go/marketbyorder-bot/coordinator.go | 37 ++++--- go/marketbyorder-bot/coordinator_test.go | 121 +++++++++++++++++++++-- go/marketbyorder-bot/shard.go | 36 +++++-- 3 files changed, 167 insertions(+), 27 deletions(-) diff --git a/go/marketbyorder-bot/coordinator.go b/go/marketbyorder-bot/coordinator.go index 374b3fc..3bcf6a0 100644 --- a/go/marketbyorder-bot/coordinator.go +++ b/go/marketbyorder-bot/coordinator.go @@ -6,7 +6,7 @@ import "context" // and routes each record to exactly one shard (by instrument_id % N), or to a // direct-write / barrier / fence path. Shards own all instrument-scoped state. // -// Dispatch is NOT safe for concurrent callers: it mutates resetSeen/resetCount/ +// Dispatch is NOT safe for concurrent callers: it mutates resetCount/ // snapshotRoute/seqLast/manifest without locks, on the assumption that the // only caller is the synchronous bot read loop. type Coordinator struct { @@ -16,9 +16,13 @@ type Coordinator struct { eventsW *EventsWriter metrics *Metrics - resetSeen bool - resetCount uint8 - manifest ManifestState // parity bookkeeping; not read for logic + // Reset Count is per publisher, and a group can carry two redundant + // publishers interleaved on the same ports under different channel_ids. + // Held as one global value, their differing-but-steady counts read as a + // reset on every alternation between them, wiping instrument state + // faster than it could be relearned. + resetCount map[uint8]uint8 // per channel_id + manifest ManifestState // parity bookkeeping; not read for logic seqLast map[string]uint64 snapshotRoute map[snapKey]int } @@ -38,19 +42,19 @@ func NewCoordinator(ctx context.Context, shards []*Shard, eventsW *EventsWriter, metrics: metrics, seqLast: map[string]uint64{}, snapshotRoute: map[snapKey]int{}, + + resetCount: map[uint8]uint8{}, } } // Dispatch implements Dispatcher. Called synchronously from the bot read loop. func (c *Coordinator) Dispatch(rec Record) { // Channel-reset barrier: reset_count change. (Implemented in Task 7.) - if c.resetSeen && rec.ResetCount != c.resetCount { + if prev, seen := c.resetCount[rec.ChannelID]; seen && rec.ResetCount != prev { c.runResetBarrier(rec) return - } - if !c.resetSeen { - c.resetSeen = true - c.resetCount = rec.ResetCount + } else if !seen { + c.resetCount[rec.ChannelID] = rec.ResetCount } c.seqLast[rec.Port] = rec.SequenceNumber @@ -107,11 +111,12 @@ func recPtr(rec Record) *Record { // (the bot is shutting down), we abandon the barrier and return without // routing the held record. No consistency requirement to uphold post-shutdown. func (c *Coordinator) runResetBarrier(held Record) { + ch := held.ChannelID acks := make(chan int, c.n) for _, s := range c.shards { go func(s *Shard) { select { - case s.inbox <- shardMsg{kind: msgReset, ack: acks}: + case s.inbox <- shardMsg{kind: msgReset, ch: ch, ack: acks}: case <-c.ctx.Done(): } }(s) @@ -127,14 +132,18 @@ func (c *Coordinator) runResetBarrier(held Record) { if c.metrics != nil { c.metrics.ChannelResetsTotal.Inc() } - c.snapshotRoute = map[snapKey]int{} + for k := range c.snapshotRoute { + if k.ch == ch { + delete(c.snapshotRoute, k) + } + } c.seqLast = map[string]uint64{} c.manifest = ManifestState{} - c.resetCount = held.ResetCount + c.resetCount[ch] = held.ResetCount // Route the held record as the first new-era frame, via the full classifier. - // resetSeen is already true and resetCount now equals held.ResetCount, so - // this re-entry into Dispatch falls through to normal classification. + // resetCount[ch] now equals held.ResetCount, so this re-entry into Dispatch + // falls through to normal classification. c.Dispatch(held) } diff --git a/go/marketbyorder-bot/coordinator_test.go b/go/marketbyorder-bot/coordinator_test.go index 140e59f..1c7bc00 100644 --- a/go/marketbyorder-bot/coordinator_test.go +++ b/go/marketbyorder-bot/coordinator_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "sync/atomic" "testing" "time" ) @@ -120,8 +121,8 @@ func TestCoordinator_ResetBarrierWipesShardsThenRoutesHeldRecord(t *testing.T) { if !newHere { t.Error("held first new-era record (instrument 5) not applied to shard 2") } - if c.resetCount != 2 { - t.Errorf("coordinator resetCount = %d, want 2", c.resetCount) + if c.resetCount[0] != 2 { + t.Errorf("coordinator resetCount[0] = %d, want 2", c.resetCount[0]) } if got := testCounter(t, metrics.ChannelResetsTotal); got != 1 { t.Errorf("channel_resets_total = %v, want 1", got) @@ -231,8 +232,8 @@ func TestCoordinator_ResetBarrierHandlesChannelScopedFirstFrame(t *testing.T) { c.Dispatch(Record{Type: "manifest_summary", ChannelID: 0, ResetCount: 2, Timestamp: time.Unix(1700000000, 0), Fields: map[string]any{ "manifest_seq": float64(1), "valid": float64(1), "instrument_count": float64(0)}}) - if c.resetCount != 2 { - t.Errorf("resetCount = %d, want 2", c.resetCount) + if c.resetCount[0] != 2 { + t.Errorf("resetCount[0] = %d, want 2", c.resetCount[0]) } } @@ -251,8 +252,7 @@ func TestCoordinator_ResetBarrierEscapesOnCtxCancel(t *testing.T) { defer cancel() c := NewCoordinator(ctx, shards, NewEventsWriter(nil), metrics) // Prime the barrier predicate. Do NOT start shard.Run, so no acks can arrive. - c.resetSeen = true - c.resetCount = 1 + c.resetCount[0] = 1 dispatchDone := make(chan struct{}) go func() { @@ -307,3 +307,112 @@ func TestCoordinator_FenceEscapesOnCtxCancel(t *testing.T) { t.Fatal("coordinator fence hung after ctx cancel") } } + +// newCoordWithShards is newCoordWithCapture's sibling, returning the shards +// themselves so a test can assert on the state a barrier did or did not wipe. +func newCoordWithShards(n int) (*Coordinator, []*Shard) { + metrics := stubMetrics() + shards := make([]*Shard, n) + for i := 0; i < n; i++ { + shards[i] = NewShard(i, n, NewEventsWriter(nil), nil, metrics) + } + return NewCoordinator(context.Background(), shards, NewEventsWriter(nil), metrics), shards +} + +// countResetBarriers dispatches recs and reports how many reset barriers fired, +// acking each one so a barrier that does fire cannot wedge the ack wait, and +// applying the wipe the barrier orders so shard state reflects it. +func countResetBarriers(t *testing.T, c *Coordinator, shards []*Shard, recs []Record) int { + t.Helper() + var n int64 + done := make(chan struct{}) + stop := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case <-stop: + return + default: + } + for i := range shards { + select { + case m := <-shards[i].inbox: + if m.kind == msgReset { + atomic.AddInt64(&n, 1) + shards[i].resetChannel(m.ch) + m.ack <- i + } + default: + } + } + } + }() + for _, r := range recs { + c.Dispatch(r) + } + close(stop) + <-done + return int(atomic.LoadInt64(&n)) +} + +// A group can carry two redundant publishers interleaved on the same ports, +// distinguished only by channel_id. Reset Count is per publisher and is stable +// while neither is resetting, so the differing values must NOT be read as a +// reset. +func TestDispatch_InterleavedChannelsWithDistinctResetCountsRunNoBarrier(t *testing.T) { + c, shards := newCoordWithShards(2) + + var recs []Record + for i := 0; i < 8; i++ { + a := Record{Type: "trade", Port: "mktdata", ChannelID: 10, InstrumentID: 2, + ResetCount: 200, Fields: map[string]any{}} + b := Record{Type: "trade", Port: "mktdata", ChannelID: 110, InstrumentID: 2, + ResetCount: 194, Fields: map[string]any{}} + recs = append(recs, a, b) + } + + if got := countResetBarriers(t, c, shards, recs); got != 0 { + t.Errorf("interleaving two steady channels ran %d reset barriers, want 0", got) + } +} + +// A genuine Reset Count change on one channel must still run a barrier, and must +// leave the other channel's instruments, refdata and snapshot contexts intact. +func TestDispatch_ResetOnOneChannelSparesTheOther(t *testing.T) { + c, shards := newCoordWithShards(1) + s := shards[0] + + keep := instKey{ch: 110, id: 2} + wipe := instKey{ch: 10, id: 2} + s.instruments[keep] = NewInstrument(2, "KEEP", -2, -8) + s.instruments[wipe] = NewInstrument(2, "WIPE", -2, -8) + s.refdata[keep] = InstrumentDef{Symbol: "KEEP"} + s.refdata[wipe] = InstrumentDef{Symbol: "WIPE"} + s.snapCtx[snapKey{ch: 110, snap: 1}] = SnapshotContext{} + s.snapCtx[snapKey{ch: 10, snap: 1}] = SnapshotContext{} + + steady := Record{Type: "trade", Port: "mktdata", ChannelID: 10, InstrumentID: 2, + ResetCount: 200, Fields: map[string]any{}} + bumped := Record{Type: "trade", Port: "mktdata", ChannelID: 10, InstrumentID: 2, + ResetCount: 201, Fields: map[string]any{}} + + if got := countResetBarriers(t, c, shards, []Record{steady, bumped}); got != 1 { + t.Fatalf("a real Reset Count change ran %d barriers, want 1", got) + } + if _, ok := s.instruments[keep]; !ok { + t.Error("channel 110 instrument was wiped by a channel 10 reset") + } + if _, ok := s.refdata[keep]; !ok { + t.Error("channel 110 refdata was wiped by a channel 10 reset") + } + if _, ok := s.snapCtx[snapKey{ch: 110, snap: 1}]; !ok { + t.Error("channel 110 snapshot context was wiped by a channel 10 reset") + } + if _, ok := s.instruments[wipe]; ok { + t.Error("channel 10 instrument survived its own channel's reset") + } + if _, ok := s.snapCtx[snapKey{ch: 10, snap: 1}]; ok { + t.Error("channel 10 snapshot context survived its own channel's reset") + } +} diff --git a/go/marketbyorder-bot/shard.go b/go/marketbyorder-bot/shard.go index 24287ed..7cd20d9 100644 --- a/go/marketbyorder-bot/shard.go +++ b/go/marketbyorder-bot/shard.go @@ -86,11 +86,33 @@ func NewShard(idx, n int, eventsW *EventsWriter, sw *SnapshotWriter, metrics *Me } } -func (s *Shard) reset() { - s.instruments = map[instKey]*Instrument{} - s.refdata = map[instKey]InstrumentDef{} - s.deltaBuf = map[instKey][]BufferedDelta{} - s.snapCtx = map[snapKey]SnapshotContext{} +// resetChannel discards every instrument owned by one channel. +// +// Scoped to a channel, not the whole shard, because a group can carry two +// redundant publishers interleaved on the same ports under different +// channel_ids. Reset Count is per publisher, so a reset on one says nothing +// about the other, and wiping both would throw away books that never reset. +func (s *Shard) resetChannel(ch uint8) { + for k := range s.instruments { + if k.ch == ch { + delete(s.instruments, k) + } + } + for k := range s.refdata { + if k.ch == ch { + delete(s.refdata, k) + } + } + for k := range s.deltaBuf { + if k.ch == ch { + delete(s.deltaBuf, k) + } + } + for k := range s.snapCtx { + if k.ch == ch { + delete(s.snapCtx, k) + } + } } // apply mutates book state for one record and returns the resulting events. @@ -482,7 +504,7 @@ func (s *Shard) Run(ctx context.Context) { s.handle(*msg.rec) case msgReset: s.mu.Lock() - s.reset() + s.resetChannel(msg.ch) s.mu.Unlock() if s.sw != nil { s.sw.Reset(ctx) // ctx-aware: never wedges on shutdown @@ -503,10 +525,10 @@ func (s *Shard) Run(ctx context.Context) { } } -// shardMsg is the inbox protocol; populated in Task 5. type shardMsg struct { rec *Record kind shardMsgKind + ch uint8 // channel to wipe, for msgReset ack chan int } From c0616f8198471aea2875cafe55af4e7b55c7f644 Mon Sep 17 00:00:00 2001 From: Andrew McConnell Date: Mon, 10 Aug 2026 18:33:26 -0500 Subject: [PATCH 2/2] marketbyorder-bot: restore shardMsg doc comment and make reset channel explicit in test --- go/marketbyorder-bot/shard.go | 4 ++++ go/marketbyorder-bot/shard_test.go | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/go/marketbyorder-bot/shard.go b/go/marketbyorder-bot/shard.go index 7cd20d9..7375757 100644 --- a/go/marketbyorder-bot/shard.go +++ b/go/marketbyorder-bot/shard.go @@ -525,6 +525,10 @@ func (s *Shard) Run(ctx context.Context) { } } +// shardMsg is the inbox protocol. A record mutates book state; a reset wipes one +// channel's share of it and acks; a fence only acks, which is enough to order a +// channel-scoped write after every preceding instrument write because the inbox +// is FIFO. type shardMsg struct { rec *Record kind shardMsgKind diff --git a/go/marketbyorder-bot/shard_test.go b/go/marketbyorder-bot/shard_test.go index 6e92b9a..8a2fbf9 100644 --- a/go/marketbyorder-bot/shard_test.go +++ b/go/marketbyorder-bot/shard_test.go @@ -201,7 +201,7 @@ func TestShard_RunProcessesRecordsThenResetAcks(t *testing.T) { s.inbox <- shardMsg{kind: msgRecord, rec: &rec} acks := make(chan int, 1) - s.inbox <- shardMsg{kind: msgReset, ack: acks} + s.inbox <- shardMsg{kind: msgReset, ch: 0, ack: acks} select { case got := <-acks: if got != 0 {