diff --git a/internal/connector/intake.go b/internal/connector/intake.go index d881c63fe..f89cbe27f 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -285,8 +285,10 @@ func New(opts Options) (*Intake, error) { if opts.Logger == nil { opts.Logger = slog.New(slog.DiscardHandler) } - // The queue reports a callback that panicked, and a queue built by a - // caller who did not think about that would report it nowhere. It is + // The queue reports its backlog edges - the warning, the pause of the + // feed read, the resume and the recovery - and a callback that panicked, + // and a queue built by a caller who did not think about that would report + // them nowhere. It is // adopted under the queue's own lock, because the queue may already be // in use — admission takes from it — and a logger the caller did set is // kept. diff --git a/internal/connector/queue.go b/internal/connector/queue.go index 37132d094..87cff1b20 100644 --- a/internal/connector/queue.go +++ b/internal/connector/queue.go @@ -34,6 +34,17 @@ const ( type Queue struct { ids chan int64 warnAt int + // recoverAt and resumeAt are the depths at or below which a warning and a + // pause are reported as over: half of the depth that raised each. An edge + // that cleared one item below where it was raised would be reported twice + // for every event of a backlog that hovers at its threshold, and a pause + // at capacity would be reported twice for every event the feed delivers, + // since each take lets one waiting offer in and the next offer waits + // again. Half is a band that a hovering backlog cannot cross by itself, + // it needs no clock, and both bands are zero or more, so a drained queue + // always ends recovered and resumed. + recoverAt int + resumeAt int // edges serializes the depth sample with the warning transition and its // callback. Unserialized, an offer can sample a warning depth, a take can @@ -58,24 +69,32 @@ type Queue struct { // finished last leaves the operator reading "resumed" while the feed is // paused. waiting int - paused bool + // paused is the reported pause, not the live one. It is raised when an + // offer first waits for room and lowered only once no offer waits and the + // depth has drained to resumeAt. Paused reports the live state. + paused bool - // OnWarn fires when the depth first crosses the warning threshold, and - // OnRecover when it falls back below. Both are optional. + // OnWarn fires when the depth reaches the warning threshold, and + // OnRecover when it has drained to half of it. Both are optional, and + // neither is needed for the operator to hear of it: every edge is + // reported on the logger first. OnWarn func(depth int) OnRecover func(depth int) - // OnPause fires when an offer begins waiting for room, and OnResume when - // it stops. The feed is not being consumed in between. + // OnPause fires when an offer first waits for room, with the depth the + // queue holds, not counting the waiting offer. OnResume fires once no + // offer waits and the depth has drained to half the pause threshold. In + // between, the feed is read only as fast as admission makes room. OnPause func(depth int) OnResume func(depth int) - // log reports a callback that panicked. Held under edges: a queue can be - // adopted by intake while another component is already using it, and a - // field written by one and read by the other is a race. + // log reports every edge and any callback that panicked. Held under + // edges: a queue can be adopted by intake while another component is + // already using it, and a field written by one and read by the other is a + // race. log *slog.Logger } -// SetLogger sets where a callback that panicked is reported. Safe to call -// while the queue is in use. +// SetLogger sets where the backlog's edges, and a callback that panicked, are +// reported. Safe to call while the queue is in use. func (q *Queue) SetLogger(log *slog.Logger) { q.edges.Lock() defer q.edges.Unlock() @@ -103,7 +122,12 @@ func NewQueue(warnAt, pauseAt int) (*Queue, error) { // Capacity IS the pause threshold: a full channel is a blocked offer is a // feed that has stopped being read. There is no second mechanism to keep // in agreement with this one. - return &Queue{ids: make(chan int64, pauseAt), warnAt: warnAt}, nil + return &Queue{ + ids: make(chan int64, pauseAt), + warnAt: warnAt, + recoverAt: warnAt / 2, + resumeAt: pauseAt / 2, + }, nil } // Offer hands an event id to admission, waiting for room when the backlog is @@ -223,15 +247,7 @@ func (q *Queue) stage(delta int) { q.edges.Lock() defer q.edges.Unlock() q.depth += delta - depth := q.depth - switch { - case depth >= q.warnAt && !q.warned: - q.warned = true - q.pending = append(q.pending, queueEdge{fire: q.OnWarn, depth: depth}) - case depth < q.warnAt && q.warned: - q.warned = false - q.pending = append(q.pending, queueEdge{fire: q.OnRecover, depth: depth}) - } + q.decide() } // stageWait records an offer starting or finishing its wait for room and the @@ -247,13 +263,36 @@ func (q *Queue) stageWait(delta int) { q.edges.Lock() defer q.edges.Unlock() q.waiting += delta - switch { - case q.waiting > 0 && !q.paused: + q.decide() +} + +// decide stages every edge the current depth and waiters cross. It is called +// holding edges, after any change to either. +// +// A raised edge clears only at its band, not one item below where it was +// raised, so a backlog that hovers at a threshold is one report each way. +// The edges up are decided before the edges down. One change can clear both +// the pause and the warning where their bands meet, and then the resume is +// told before the recovery, the reverse of the order they were raised in. +func (q *Queue) decide() { + if !q.warned && q.depth >= q.warnAt { + q.warned = true + q.pending = append(q.pending, queueEdge{kind: edgeWarn, fire: q.OnWarn, depth: q.depth}) + } + if !q.paused && q.waiting > 0 { q.paused = true - q.pending = append(q.pending, queueEdge{fire: q.OnPause, depth: q.depth}) - case q.waiting == 0 && q.paused: + // The nonblocking send observed capacity. Other offers may have + // counted their ids without registering as waiters yet, so the + // independently staged counters cannot tell us that held depth. + q.pending = append(q.pending, queueEdge{kind: edgePause, fire: q.OnPause, depth: cap(q.ids)}) + } + if q.paused && q.waiting == 0 && q.depth <= q.resumeAt { q.paused = false - q.pending = append(q.pending, queueEdge{fire: q.OnResume, depth: q.depth}) + q.pending = append(q.pending, queueEdge{kind: edgeResume, fire: q.OnResume, depth: q.depth}) + } + if q.warned && q.depth <= q.recoverAt { + q.warned = false + q.pending = append(q.pending, queueEdge{kind: edgeRecover, fire: q.OnRecover, depth: q.depth}) } } @@ -290,7 +329,12 @@ func (q *Queue) deliver() { } } -// fire runs one callback with the lock released and a panic contained. +// fire reports one edge and runs its callback, with the lock released and a +// panic contained. +// +// The report comes first and does not depend on the callback: a caller that +// set none, as production does, still has the backlog reported rather than +// hidden, and one that panics cannot take the report down with it. // // The callbacks belong to whoever built the queue, and the goroutine they run // on is the feed's delivery path or a worker taking work off it. A callback @@ -299,9 +343,11 @@ func (q *Queue) deliver() { // callback is told about was committed before it ran, so what it does cannot // change it. func (q *Queue) fire(edge queueEdge) { + log := q.logger() + q.report(log, edge) defer func() { if p := recover(); p != nil { - q.logger().Error("a backlog callback panicked; the queue carried on without it", + log.Error("a backlog callback panicked; the queue carried on without it", "panic", richtext.SanitizeSingleLine(fmt.Sprint(p))) } }() @@ -310,6 +356,30 @@ func (q *Queue) fire(edge queueEdge) { } } +// backlogNextStep is what a warning tells the operator to look at. +const backlogNextStep = "The usual cause is slow or failing Basecamp reads during admission; basecamp connect status shows the backlog of records by state." + +// report tells the operator about one edge, with the depth it was decided at, +// the threshold it crossed and the depth at which it clears. The depth of a +// pause is what the queue holds; the offer waiting for room is not counted. +func (q *Queue) report(log *slog.Logger, edge queueEdge) { + pauseAt := cap(q.ids) + switch edge.kind { + case edgeWarn: + log.Warn("the backlog reached its warning depth; admission is falling behind the feed. "+backlogNextStep, + "depth", edge.depth, "warn_at", q.warnAt, "recover_at", q.recoverAt, "pause_at", pauseAt) + case edgeRecover: + log.Info("the backlog fell back below its warning depth", + "depth", edge.depth, "warn_at", q.warnAt, "recover_at", q.recoverAt) + case edgePause: + log.Warn("the backlog reached its pause depth; at least one producer is waiting for room in the queue. "+backlogNextStep, + "depth", edge.depth, "pause_at", pauseAt, "resume_at", q.resumeAt) + case edgeResume: + log.Info("the backlog drained to half its pause depth; no producers are waiting for room", + "depth", edge.depth, "pause_at", pauseAt, "resume_at", q.resumeAt) + } +} + // logger is read under edges, so it must never be called holding them. Its one // caller, fire, runs with the lock released. func (q *Queue) logger() *slog.Logger { @@ -322,6 +392,17 @@ func (q *Queue) logger() *slog.Logger { } type queueEdge struct { + kind edgeKind fire func(int) depth int } + +// edgeKind is which of the backlog's transitions an edge is. +type edgeKind int + +const ( + edgeWarn edgeKind = iota + edgeRecover + edgePause + edgeResume +) diff --git a/internal/connector/queue_test.go b/internal/connector/queue_test.go index 8a764b063..fa8617d4a 100644 --- a/internal/connector/queue_test.go +++ b/internal/connector/queue_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "log/slog" + "strings" "sync" "sync/atomic" "testing" @@ -57,9 +58,9 @@ func TestQueuePausesTheCallerAtTheThreshold(t *testing.T) { select { case depth := <-paused: - // Three: the two in the queue and the one waiting to go in, which is - // backlog too. - assert.Equal(t, 3, depth) + // Two: what the queue holds. The offer waiting to go in is not + // counted, so the depth reported is never above the threshold. + assert.Equal(t, 2, depth) case <-time.After(2 * time.Second): t.Fatal("the third offer should have waited for room") } @@ -220,34 +221,45 @@ func TestPauseAndResumeAreObservedInTheOrderTheyHappened(t *testing.T) { waiters.Add(1) go func() { defer waiters.Done() - assert.NoError(t, queue.Offer(ctx, 2)) // waits for room, then resumes + assert.NoError(t, queue.Offer(ctx, 2)) // waits for room }() require.Eventually(t, queue.Paused, time.Second, time.Millisecond) first, err := queue.Take(ctx) require.NoError(t, err) assert.Equal(t, int64(1), first) + waiters.Wait() + // The take that drains the queue to its resume band delivers the resume, + // and the callback holds it there. + drained := make(chan int64, 1) + go func() { + id, err := queue.Take(ctx) + assert.NoError(t, err) + drained <- id + }() <-resumeStarted // the resume has begun and is not finished + require.NoError(t, queue.Offer(ctx, 3)) waiters.Add(1) go func() { defer waiters.Done() - assert.NoError(t, queue.Offer(ctx, 3)) // pauses again, mid-resume + assert.NoError(t, queue.Offer(ctx, 4)) // pauses again, mid-resume }() require.Eventually(t, queue.Paused, time.Second, time.Millisecond) close(holdResume) + assert.Equal(t, int64(2), <-drained) require.Eventually(t, func() bool { return len(observed()) == 3 }, time.Second, time.Millisecond) assert.Equal(t, []string{"paused", "resumed", "paused"}, observed(), "a resume that started first must not be reported after the pause that followed it") - second, err := queue.Take(ctx) - require.NoError(t, err) - assert.Equal(t, int64(2), second) - waiters.Wait() third, err := queue.Take(ctx) require.NoError(t, err) assert.Equal(t, int64(3), third) + waiters.Wait() + fourth, err := queue.Take(ctx) + require.NoError(t, err) + assert.Equal(t, int64(4), fourth) assert.False(t, queue.Paused()) } @@ -284,10 +296,12 @@ func TestAPanickingPauseCallbackLeavesNoPhantomBacklog(t *testing.T) { assert.False(t, queue.Paused(), "no phantom pause") assert.Contains(t, logs.String(), "a backlog callback panicked") - // And the next crossing is still reported, on both sides. + // And the next crossing is still reported, on both sides. The resume is + // reported once the queue drains to its band, here empty. second, err := queue.Take(ctx) require.NoError(t, err) assert.Equal(t, int64(2), second) + assert.Equal(t, int32(1), resumes.Load()) require.NoError(t, queue.Offer(ctx, 3)) blocked := make(chan error, 1) go func() { blocked <- queue.Offer(ctx, 4) }() @@ -295,7 +309,138 @@ func TestAPanickingPauseCallbackLeavesNoPhantomBacklog(t *testing.T) { _, err = queue.Take(ctx) require.NoError(t, err) require.NoError(t, <-blocked) + _, err = queue.Take(ctx) + require.NoError(t, err) assert.Equal(t, int32(2), pauses.Load()) assert.Equal(t, int32(2), resumes.Load()) - assert.Equal(t, 1, queue.Depth()) + assert.Zero(t, queue.Depth()) +} + +// The report is the queue's, not the callback's: a callback that panics is +// contained after the edge has already been reported. +func TestAPanickingCallbackDoesNotSilenceTheReport(t *testing.T) { + queue, err := NewQueue(1, 4) + require.NoError(t, err) + var logs bytes.Buffer + queue.SetLogger(slog.New(slog.NewTextHandler(&logs, nil))) + queue.OnWarn = func(int) { + assert.Contains(t, logs.String(), "the backlog reached its warning depth") + panic("a warning callback panics") + } + + require.NoError(t, queue.Offer(context.Background(), 1)) + + assert.Contains(t, logs.String(), "the backlog reached its warning depth") + assert.Contains(t, logs.String(), "depth=1 warn_at=1 recover_at=0 pause_at=4") + assert.Contains(t, logs.String(), "a backlog callback panicked") +} + +// A backlog that hovers at a threshold is one report each way, not one per +// event. At capacity each take lets the waiting offer in and the next offer +// waits again; that is still one pause, and the resume is reported only once +// the backlog has drained to half the pause depth. The warning clears the +// same way, at half the warning depth. +func TestAPauseDepthExcludesAnOfferNotYetRegisteredAsWaiting(t *testing.T) { + queue, err := NewQueue(2, 3) + require.NoError(t, err) + for id := range int64(3) { + require.NoError(t, queue.Offer(t.Context(), id)) + } + // An offer preempted between stage(1) and stageWait(1) has counted its + // id but is not held by the full channel. A second offer must report + // capacity, not include that first offer in the held depth. + queue.stage(1) + paused := make(chan int, 1) + queue.OnPause = func(depth int) { paused <- depth } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { done <- queue.Offer(ctx, 4) }() + select { + case depth := <-paused: + assert.Equal(t, 3, depth) + case <-time.After(time.Second): + t.Fatal("the second offer did not reach the full queue") + } + cancel() + require.ErrorIs(t, <-done, context.Canceled) + queue.stage(-1) // The preempted first offer is canceled too. + for range 3 { + _, err := queue.Take(t.Context()) + require.NoError(t, err) + } + assert.Zero(t, queue.Depth()) +} + +func TestABacklogHoveringAtItsThresholdsIsReportedOnceEachWay(t *testing.T) { + queue, err := NewQueue(4, 8) + require.NoError(t, err) + var logs safeBuffer + queue.SetLogger(slog.New(slog.NewTextHandler(&logs, nil))) + var mu sync.Mutex + var seen []string + record := func(what string) func(int) { + return func(int) { mu.Lock(); seen = append(seen, what); mu.Unlock() } + } + queue.OnWarn = record("warn") + queue.OnRecover = record("recover") + queue.OnPause = record("pause") + queue.OnResume = record("resume") + observed := func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), seen...) + } + ctx := context.Background() + take := func() { + t.Helper() + _, err := queue.Take(ctx) + require.NoError(t, err) + } + + // Around the warning depth: 3, 4, 3, 4, ... + for i := range 4 { + require.NoError(t, queue.Offer(ctx, int64(i))) + } + for i := range 20 { + take() + require.NoError(t, queue.Offer(ctx, int64(100+i))) + } + assert.Equal(t, []string{"warn"}, observed()) + + // Around the pause depth: full, one waits, one taken, the next waits. + for i := range 4 { + require.NoError(t, queue.Offer(ctx, int64(200+i))) + } + for i := range 20 { + waiting := make(chan error, 1) + go func() { waiting <- queue.Offer(ctx, int64(300+i)) }() + require.Eventually(t, queue.Paused, time.Second, time.Millisecond) + take() + require.NoError(t, <-waiting) + } + assert.Equal(t, []string{"warn", "pause"}, observed()) + + // Draining: nothing until half the pause depth, then the resume; nothing + // more until half the warning depth, then the recovery. + for range 3 { + take() + } + assert.Equal(t, []string{"warn", "pause"}, observed(), "depth 5 is above the resume band") + take() + assert.Equal(t, []string{"warn", "pause", "resume"}, observed()) + take() + assert.Equal(t, []string{"warn", "pause", "resume"}, observed(), "depth 3 is above the recovery band") + take() + assert.Equal(t, []string{"warn", "pause", "resume", "recover"}, observed()) + assert.Equal(t, 2, queue.Depth()) + + text := logs.String() + assert.Equal(t, 1, strings.Count(text, "the backlog reached its warning depth")) + assert.Equal(t, 1, strings.Count(text, "the backlog reached its pause depth")) + assert.Equal(t, 1, strings.Count(text, "the backlog drained to half its pause depth")) + assert.Equal(t, 1, strings.Count(text, "the backlog fell back below its warning depth")) + assert.Contains(t, text, "depth=8 pause_at=8 resume_at=4", "the pause reports what the queue holds") + assert.Contains(t, text, "depth=4 pause_at=8 resume_at=4") + assert.Contains(t, text, "depth=2 warn_at=4 recover_at=2") } diff --git a/internal/connector/review2_test.go b/internal/connector/review2_test.go index 619e76d6d..1c92598f0 100644 --- a/internal/connector/review2_test.go +++ b/internal/connector/review2_test.go @@ -244,12 +244,17 @@ func TestQueuePauseTracksEveryBlockedOffer(t *testing.T) { t.Fatal("one offer should have resumed") } assert.True(t, queue.Paused(), "the other offer is still waiting, so the feed is still paused") - assert.Zero(t, resumes.Load(), "resume fires when the last waiter stops waiting") + assert.Zero(t, resumes.Load(), "no resume while an offer still waits") _, err = queue.Take(ctx) require.NoError(t, err) require.Eventually(t, func() bool { return !queue.Paused() }, time.Second, time.Millisecond) + assert.Zero(t, resumes.Load(), "no resume until the backlog drains to its band") + + _, err = queue.Take(ctx) + require.NoError(t, err) assert.Equal(t, int32(1), pauses.Load()) + require.Eventually(t, func() bool { return resumes.Load() == 1 }, time.Second, time.Millisecond) assert.Equal(t, int32(1), resumes.Load()) } diff --git a/internal/connector/seam_contract_test.go b/internal/connector/seam_contract_test.go index a41d9ebf0..1702c161d 100644 --- a/internal/connector/seam_contract_test.go +++ b/internal/connector/seam_contract_test.go @@ -3,6 +3,7 @@ package connector import ( "bytes" "context" + "encoding/json" "errors" "log/slog" "net/http" @@ -11,6 +12,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -202,6 +204,94 @@ func TestIntakeGivesTheQueueItsLogger(t *testing.T) { assert.Contains(t, logs.String(), "a backlog callback panicked") } +// Production builds its queue with no callbacks and hands it to intake with +// the connector's logger, so the logger is the only place an operator can +// learn that the backlog is growing or that the feed read is held back. Each +// edge is reported there, with the depth, the threshold it crossed and the +// band at which it clears. +func TestIntakeReportsTheBacklogsEdgesOnItsLogger(t *testing.T) { + var logs safeBuffer + queue, err := NewQueue(2, 3) + require.NoError(t, err) + _, err = New(Options{ + Origin: "https://3.basecampapi.com", + AccountID: "2914079", + ConsumerNamespace: "connector-test", + Ledger: newTestLedger(t), + Queue: queue, + Minter: stubMinter{}, + Polls: &scriptedPolls{}, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + }) + require.NoError(t, err) + + ctx := context.Background() + require.NoError(t, queue.Offer(ctx, 1)) + require.NoError(t, queue.Offer(ctx, 2)) + require.NoError(t, queue.Offer(ctx, 3)) + waiting := make(chan error, 1) + go func() { waiting <- queue.Offer(ctx, 4) }() + require.Eventually(t, func() bool { return len(backlogLines(t, logs.String())) == 2 }, time.Second, time.Millisecond) + // One take makes room and the waiting offer goes in, but the backlog is + // still at its pause depth, so nothing is reported yet. + _, err = queue.Take(ctx) + require.NoError(t, err) + require.NoError(t, <-waiting) + _, err = queue.Take(ctx) + require.NoError(t, err) + assert.Len(t, backlogLines(t, logs.String()), 2) + // The take to depth 1 reaches both bands, and the resume is told first. + _, err = queue.Take(ctx) + require.NoError(t, err) + + lines := backlogLines(t, logs.String()) + require.Len(t, lines, 4) + + assert.Equal(t, "WARN", lines[0]["level"]) + assert.Contains(t, lines[0]["msg"], "warning depth") + assert.EqualValues(t, 2, lines[0]["depth"]) + assert.EqualValues(t, 2, lines[0]["warn_at"]) + assert.EqualValues(t, 1, lines[0]["recover_at"]) + assert.EqualValues(t, 3, lines[0]["pause_at"]) + assert.Contains(t, lines[0]["msg"], "basecamp connect status") + + assert.Equal(t, "WARN", lines[1]["level"]) + assert.Contains(t, lines[1]["msg"], "pause depth") + assert.Contains(t, lines[1]["msg"], "basecamp connect status") + assert.EqualValues(t, 3, lines[1]["depth"], "what the queue holds, not the offer waiting") + assert.EqualValues(t, 3, lines[1]["pause_at"]) + assert.EqualValues(t, 1, lines[1]["resume_at"]) + + assert.Equal(t, "INFO", lines[2]["level"]) + assert.Contains(t, lines[2]["msg"], "no producers are waiting") + assert.EqualValues(t, 1, lines[2]["depth"]) + assert.EqualValues(t, 3, lines[2]["pause_at"]) + assert.EqualValues(t, 1, lines[2]["resume_at"]) + + assert.Equal(t, "INFO", lines[3]["level"]) + assert.Contains(t, lines[3]["msg"], "below its warning depth") + assert.EqualValues(t, 1, lines[3]["depth"]) + assert.EqualValues(t, 2, lines[3]["warn_at"]) + assert.EqualValues(t, 1, lines[3]["recover_at"]) +} + +// backlogLines decodes the JSON log lines that report a backlog edge. +func backlogLines(t *testing.T, logs string) []map[string]any { + t.Helper() + var lines []map[string]any + for line := range strings.SplitSeq(strings.TrimSpace(logs), "\n") { + if line == "" { + continue + } + var record map[string]any + require.NoError(t, json.Unmarshal([]byte(line), &record)) + if msg, _ := record["msg"].(string); strings.HasPrefix(msg, "the backlog ") { + lines = append(lines, record) + } + } + return lines +} + // A queue can already be in use when intake adopts it — admission takes from // the same queue — so adopting its logger must not race the callback path that // reads it.