diff --git a/internal/link/calls.go b/internal/link/calls.go index 6332c705..c91512e6 100644 --- a/internal/link/calls.go +++ b/internal/link/calls.go @@ -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) } } diff --git a/internal/link/calls_test.go b/internal/link/calls_test.go index 577f7c22..2b77411e 100644 --- a/internal/link/calls_test.go +++ b/internal/link/calls_test.go @@ -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") + } +} diff --git a/internal/link/meshcli_test.go b/internal/link/meshcli_test.go index 7e5db365..c2077e10 100644 --- a/internal/link/meshcli_test.go +++ b/internal/link/meshcli_test.go @@ -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) }