Queue one state per call for the bus, bounded far above a burst, so a burst of read calls no longer drops call records (issue 774)

At 17:44:53 the controller said 30 calls (tools, nodes, modules) were "kept in memory only: the bus is not taking calls' records". No write had failed (no "could not be kept" line): a burst of about 200 read calls in one second queued a begin and a finish each into a channel of 256 while the one writer put them one by one, and the channel overflowed. The queue now holds the latest state of each waiting call (a finish replaces its own waiting begin, as only the last state is kept anyway), in first-queued order, bounded at 8192 calls; the writer takes the whole queue at each wake. A burst now costs one write per call and drops nothing. Quick fix: a larger channel. Thorough fix: coalesce by call, so the writes a burst asks are halved and the bound is spent on calls, not states. Chose thorough; both are small. Research 072 piece: §2's "read calls said" row: the calls bucket is the controller's own state on the bus, kept from events (each call's change), never re-read.
2026-10-30 18:21:58 +01:00
parent c421629f6e
commit 519434241a
3 changed files with 132 additions and 44 deletions
@@ -127,9 +127,13 @@ type CallLog struct {

 	// keeper keeps every call beyond this process, when Durably was given one; writes go through
 	// one goroutine, in order, so a call's last state is the one kept.
-	keeper  CallKeeper
-	holder  string
-	writes  chan Call
+	keeper CallKeeper
+	holder string
+	// queued are the calls' states waiting for the keeper, one per call, in the order each call was first queued, under
+	// qmu (keep is called with mu held); woken is told when one is queued (callQueue).
+	qmu     sync.Mutex
+	queued  callQueue
+	woken   chan struct{}
 	logger  *log.Logger
 	lost    int // writes the keeper could not take, said once each
 	keepErr error
@@ -165,8 +169,8 @@ const keepTries = 5
 func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, logger *log.Logger) error {
 	l.mu.Lock()
 	l.keeper, l.holder, l.logger = keeper, holder, logger
-	if l.writes == nil {
-		l.writes = make(chan Call, 256)
+	if l.woken == nil {
+		l.woken = make(chan struct{}, 1)
 		go l.keepWrites()
 	}
 	l.mu.Unlock()
@@ -203,51 +207,109 @@ func orSomebody(s string) string {
 	return s
 }

+// callQueueBound is how many calls may wait for the keeper at once: far beyond a burst of reads (about 200 calls in one
+// second on 2026-10-30), and a bound, so a keeper that takes nothing for long is said rather than grown into.
+const callQueueBound = 8192
+
+// callQueue holds the latest state of each call waiting for the keeper. **One state per call** (sweep 2026-10-30, issue
+// 774): a call's later state replaces its earlier one still waiting, since only the last is kept anyway; a burst of
+// read calls queued begin and finish each, and a channel of 256 dropped 30 of them in one second.
+type callQueue struct {
+	order []string
+	byID  map[string]Call
+}
+
+// put queues c, replacing a waiting state of the same call; false when the queue is full.
+func (q *callQueue) put(c Call) bool {
+	if q.byID == nil {
+		q.byID = map[string]Call{}
+	}
+	if _, waiting := q.byID[c.ID]; waiting {
+		q.byID[c.ID] = c
+		return true
+	}
+	if len(q.order) >= callQueueBound {
+		return false
+	}
+	q.order = append(q.order, c.ID)
+	q.byID[c.ID] = c
+	return true
+}
+
+// take is every waiting state, in order, and empties the queue.
+func (q *callQueue) take() []Call {
+	out := make([]Call, 0, len(q.order))
+	for _, id := range q.order {
+		out = append(out, q.byID[id])
+	}
+	q.order, q.byID = nil, nil
+	return out
+}
+
 // keep queues one call's state for the keeper. Never blocks a call: a queue that is full is a keeper
 // that is not taking writes, and that is said rather than waited on.
 func (l *CallLog) keep(c Call) {
-	if l.writes == nil {
+	if l.woken == nil {
+		return
+	}
+	l.qmu.Lock()
+	queued := l.queued.put(c)
+	if !queued {
+		l.lost++
+	}
+	lost := l.lost
+	l.qmu.Unlock()
+	if !queued {
+		if l.logger != nil {
+			l.logger.Printf("%s (%s.%s) is kept in memory only: %d calls wait for the bus to take their records "+
+				"(%d not kept)", c.ID, c.Seat, c.Verb, callQueueBound, lost)
+		}
 		return
 	}
 	select {
-	case l.writes <- c:
+	case l.woken <- struct{}{}:
 	default:
-		l.lost++
-		if l.logger != nil {
-			l.logger.Printf("%s (%s.%s) is kept in memory only: the bus is not taking calls' records (%d not kept)",
-				c.ID, c.Seat, c.Verb, l.lost)
-		}
 	}
 }

 func (l *CallLog) keepWrites() {
-	for c := range l.writes {
-		l.mu.Lock()
-		gate := l.epoch
-		l.mu.Unlock()
-		if gate != nil {
-			if _, err := gate(); err != nil {
-				if l.logger != nil {
-					l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+
-						"marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err)
-				}
-				continue
+	for range l.woken {
+		l.qmu.Lock()
+		waiting := l.queued.take()
+		l.qmu.Unlock()
+		for _, c := range waiting {
+			l.keepOne(c)
+		}
+	}
+}
+
+// keepOne writes one call's state through the keeper, under the lease.
+func (l *CallLog) keepOne(c Call) {
+	l.mu.Lock()
+	gate := l.epoch
+	l.mu.Unlock()
+	if gate != nil {
+		if _, err := gate(); err != nil {
+			if l.logger != nil {
+				l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+
+					"marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err)
 			}
+			return
 		}
-		var err error
-		for try := 0; try < keepTries; try++ {
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			err = l.keeper.Keep(ctx, c)
-			cancel()
-			if err == nil {
-				break
-			}
-			time.Sleep(time.Duration(try+1) * time.Second)
-		}
-		if err != nil && l.logger != nil {
-			l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+
-				"answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err)
+	}
+	var err error
+	for try := 0; try < keepTries; try++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		err = l.keeper.Keep(ctx, c)
+		cancel()
+		if err == nil {
+			break
 		}
+		time.Sleep(time.Duration(try+1) * time.Second)
+	}
+	if err != nil && l.logger != nil {
+		l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+
+			"answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err)
 	}
 }

@@ -254,8 +254,7 @@ func TestAHandoverRefusalIsMarkedRetryable(t *testing.T) {
 // who may call the seat, and a token is the one-time right to become a machine of the mesh.
 func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
 	l, a := NewCallLog(), newAnswers(t)
-	writes := make(chan Call, 4)
-	l.writes = writes
+	l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below
 	l.serveCall("mesh-controller", "token", json.RawMessage(`{"new":"laptop"}`), "_INBOX.x.1",
 		func(context.Context, json.RawMessage) (any, error) { return "token for laptop: s3cret-join", nil },
 		a.respond, nil)
@@ -266,8 +265,7 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {
 	if len(recent) != 1 || strings.Contains(string(recent[0].Answer), "s3cret-join") {
 		t.Fatalf("the token was kept in memory: %+v", recent)
 	}
-	close(writes)
-	for c := range writes {
+	for _, c := range l.queued.take() {
 		if strings.Contains(string(c.Answer), "s3cret-join") {
 			t.Fatalf("the token was sent to be kept on the bus: %s", c.Answer)
 		}
@@ -276,3 +274,33 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) {

 // A verb whose handler panics answers an error, and the controller serves the next call (review of
 // novox/hq issue 327: read verbs are answered in the serving process now).
+
+// Sweep 2026-10-30 (issue 774): a burst of calls never loses a record while the keeper is slow: each call waits once,
+// with its latest state, and every call of a burst far over the old queue of 256 is kept with its last state.
+func TestABurstOfCallsIsKeptWithEachCallsLastState(t *testing.T) {
+	var q callQueue
+	for i := range 2000 {
+		id := fmt.Sprintf("call-%d", i)
+		if !q.put(Call{ID: id, State: CallRunning}) || !q.put(Call{ID: id, State: CallAnswered}) {
+			t.Fatalf("%s was refused by a queue of %d", id, callQueueBound)
+		}
+	}
+	got := q.take()
+	if len(got) != 2000 {
+		t.Fatalf("%d calls wait, want 2000", len(got))
+	}
+	for i, c := range got {
+		if c.ID != fmt.Sprintf("call-%d", i) || c.State != CallAnswered {
+			t.Fatalf("call %d waits as %+v", i, c)
+		}
+	}
+	if len(q.take()) != 0 {
+		t.Fatal("taken calls wait again")
+	}
+	for i := range callQueueBound {
+		q.put(Call{ID: fmt.Sprint(i)})
+	}
+	if q.put(Call{ID: "one too many"}) {
+		t.Fatal("the queue grew past its bound")
+	}
+}
@@ -125,8 +125,7 @@ func askCLI(t *testing.T, conn *nats.Conn, node string, asked CLIAsked) (CLIAnsw
 // bus, where `calls` answers anyone who may call the seat.
 func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
 	l, a := NewCallLog(), newAnswers(t)
-	writes := make(chan Call, 4)
-	l.writes = writes
+	l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below
 	asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"})
 	l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv",
 		func(context.Context, json.RawMessage) (any, error) {
@@ -134,8 +133,7 @@ func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
 		},
 		a.respond, nil)
 	_ = a.only()
-	close(writes)
-	for c := range writes {
+	for _, c := range l.queued.take() {
 		if carries(c.Args, "s3cret") || carries(c.Answer, "s3cret-join") {
 			t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer)
 		}