marketbyorder-bot: track Reset Count per channel

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)
 }
 

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")
+	}
+}

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,14 @@ func (s *Shard) Run(ctx context.Context) {
 	}
 }
 
-// shardMsg is the inbox protocol; populated in Task 5.
+// 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
+	ch   uint8 // channel to wipe, for msgReset
 	ack  chan int
 }
 

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 {