From f194a0595aebebbc65471c8ad01e44195a9f4851 Mon Sep 17 00:00:00 2001
From: Rob Zolkos <rob@zolkos.com>
Date: Fri, 2 Oct 2026 13:15:53 -0400
Subject: [PATCH 1/3] Report when the connector's backlog warns, pauses and
 recovers

The queue between intake and admission decides four edges: the backlog
crossing its warning depth, an offer waiting for room (which stops the
feed read), that wait ending, and the backlog falling back below the
warning depth. It handed them only to optional callbacks, and
runConnect sets none, so in production every edge went nowhere. The
package says the pause is reported rather than hidden; it was hidden.

The queue now reports each edge on its logger before running any
callback, with the depth and the threshold crossed, and says plainly
when reading the feed is paused and why. Intake already adopts the
connector's logger for the queue, so every caller that builds intake
gets the report and none can forget to wire it. A callback that panics
is contained after the report, so it cannot silence it.
---
 internal/connector/intake.go             |  6 +-
 internal/connector/queue.go              | 66 +++++++++++++++----
 internal/connector/queue_test.go         | 16 +++++
 internal/connector/seam_contract_test.go | 81 ++++++++++++++++++++++++
 4 files changed, 155 insertions(+), 14 deletions(-)

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..0b7d0056c 100644
--- a/internal/connector/queue.go
+++ b/internal/connector/queue.go
@@ -61,21 +61,24 @@ type Queue struct {
 	paused  bool
 
 	// OnWarn fires when the depth first crosses the warning threshold, and
-	// OnRecover when it falls back below. Both are optional.
+	// OnRecover when it falls back below. 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  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()
@@ -227,10 +230,10 @@ func (q *Queue) stage(delta int) {
 	switch {
 	case depth >= q.warnAt && !q.warned:
 		q.warned = true
-		q.pending = append(q.pending, queueEdge{fire: q.OnWarn, depth: depth})
+		q.pending = append(q.pending, queueEdge{kind: edgeWarn, 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.pending = append(q.pending, queueEdge{kind: edgeRecover, fire: q.OnRecover, depth: depth})
 	}
 }
 
@@ -250,10 +253,10 @@ func (q *Queue) stageWait(delta int) {
 	switch {
 	case q.waiting > 0 && !q.paused:
 		q.paused = true
-		q.pending = append(q.pending, queueEdge{fire: q.OnPause, depth: q.depth})
+		q.pending = append(q.pending, queueEdge{kind: edgePause, fire: q.OnPause, depth: q.depth})
 	case q.waiting == 0 && q.paused:
 		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})
 	}
 }
 
@@ -290,7 +293,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 +307,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 +320,27 @@ func (q *Queue) fire(edge queueEdge) {
 	}
 }
 
+// report tells the operator about one edge, with the depth it was decided at
+// and the threshold it crossed. The depth of a pause or a resume counts the id
+// the waiting offer is holding: it is backlog either way.
+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",
+			"depth", edge.depth, "warn_at", q.warnAt, "pause_at", pauseAt)
+	case edgeRecover:
+		log.Info("the backlog fell back below its warning depth",
+			"depth", edge.depth, "warn_at", q.warnAt)
+	case edgePause:
+		log.Warn("the backlog reached its pause depth; reading the feed is paused until admission makes room, so the backlog cannot outgrow memory",
+			"depth", edge.depth, "pause_at", pauseAt)
+	case edgeResume:
+		log.Info("the backlog has room again; reading the feed resumed",
+			"depth", edge.depth, "pause_at", pauseAt)
+	}
+}
+
 // 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 +353,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..d570b474c 100644
--- a/internal/connector/queue_test.go
+++ b/internal/connector/queue_test.go
@@ -299,3 +299,19 @@ func TestAPanickingPauseCallbackLeavesNoPhantomBacklog(t *testing.T) {
 	assert.Equal(t, int32(2), resumes.Load())
 	assert.Equal(t, 1, 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) { 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 pause_at=4")
+	assert.Contains(t, logs.String(), "a backlog callback panicked")
+}
diff --git a/internal/connector/seam_contract_test.go b/internal/connector/seam_contract_test.go
index a41d9ebf0..127e46c1c 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,85 @@ 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 has stopped. Each
+// edge is reported there, with the depth and the threshold it crossed.
+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 is let finish before the
+	// rest, so the resume is decided before the recovery.
+	_, err = queue.Take(ctx)
+	require.NoError(t, err)
+	require.NoError(t, <-waiting)
+	for range 2 {
+		_, 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, 3, lines[0]["pause_at"])
+
+	assert.Equal(t, "WARN", lines[1]["level"])
+	assert.Contains(t, lines[1]["msg"], "reading the feed is paused")
+	assert.EqualValues(t, 3, lines[1]["pause_at"])
+	assert.Contains(t, lines[1], "depth")
+
+	assert.Equal(t, "INFO", lines[2]["level"])
+	assert.Contains(t, lines[2]["msg"], "reading the feed resumed")
+	assert.EqualValues(t, 3, lines[2]["pause_at"])
+	assert.Contains(t, lines[2], "depth")
+
+	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"])
+}
+
+// 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.

From 8b907e872b5d4fb872adc0c87efe2b676f417430 Mon Sep 17 00:00:00 2001
From: Rob Zolkos <rob@zolkos.com>
Date: Fri, 2 Oct 2026 15:22:55 -0400
Subject: [PATCH 2/3] Report each backlog edge once, with a band below the
 threshold

A backlog hovering at the warning depth wrote a warning and a recovery
for every event, and a saturated queue wrote a pause and a resume for
every event the feed delivered. A warning now clears at half the warning
depth, and a pause is reported over only once no offer waits and the
backlog has drained to half the pause depth. The feed's blocking at
capacity is unchanged; only the reported edges move.

The pause line now reports what the queue holds, not counting the offer
waiting for room, and both warning lines point at the usual cause and at
basecamp connect status.
---
 internal/connector/queue.go              | 100 +++++++++++++------
 internal/connector/queue_test.go         | 118 ++++++++++++++++++++---
 internal/connector/review2_test.go       |   6 +-
 internal/connector/seam_contract_test.go |  33 ++++---
 4 files changed, 201 insertions(+), 56 deletions(-)

diff --git a/internal/connector/queue.go b/internal/connector/queue.go
index 0b7d0056c..5895e55d8 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,16 +69,21 @@ 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, and neither is
-	// needed for the operator to hear of it: every edge is reported on the
-	// logger first.
+	// 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 every edge and any callback that panicked. Held under
@@ -106,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
@@ -226,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{kind: edgeWarn, fire: q.OnWarn, depth: depth})
-	case depth < q.warnAt && q.warned:
-		q.warned = false
-		q.pending = append(q.pending, queueEdge{kind: edgeRecover, fire: q.OnRecover, depth: depth})
-	}
+	q.decide()
 }
 
 // stageWait records an offer starting or finishing its wait for room and the
@@ -250,14 +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{kind: edgePause, fire: q.OnPause, depth: q.depth})
-	case q.waiting == 0 && q.paused:
+		// The held depth: a waiting offer has counted its id but cannot put
+		// it in, so it is not part of what the queue holds.
+		q.pending = append(q.pending, queueEdge{kind: edgePause, fire: q.OnPause, depth: q.depth - q.waiting})
+	}
+	if q.paused && q.waiting == 0 && q.depth <= q.resumeAt {
 		q.paused = false
 		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})
+	}
 }
 
 // deliver hands the staged edges to their callbacks, in the order they were
@@ -320,24 +355,27 @@ func (q *Queue) fire(edge queueEdge) {
 	}
 }
 
-// report tells the operator about one edge, with the depth it was decided at
-// and the threshold it crossed. The depth of a pause or a resume counts the id
-// the waiting offer is holding: it is backlog either way.
+// 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",
-			"depth", edge.depth, "warn_at", q.warnAt, "pause_at", pauseAt)
+		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)
+			"depth", edge.depth, "warn_at", q.warnAt, "recover_at", q.recoverAt)
 	case edgePause:
-		log.Warn("the backlog reached its pause depth; reading the feed is paused until admission makes room, so the backlog cannot outgrow memory",
-			"depth", edge.depth, "pause_at", pauseAt)
+		log.Warn("the backlog reached its pause depth; the next event waits for room, and the feed is read only as fast as admission makes room, so the backlog cannot outgrow memory. "+backlogNextStep,
+			"depth", edge.depth, "pause_at", pauseAt, "resume_at", q.resumeAt)
 	case edgeResume:
-		log.Info("the backlog has room again; reading the feed resumed",
-			"depth", edge.depth, "pause_at", pauseAt)
+		log.Info("the backlog drained to half its pause depth; the feed is read at full speed again",
+			"depth", edge.depth, "pause_at", pauseAt, "resume_at", q.resumeAt)
 	}
 }
 
diff --git a/internal/connector/queue_test.go b/internal/connector/queue_test.go
index d570b474c..277d1fca6 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,9 +309,11 @@ 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
@@ -312,6 +328,84 @@ func TestAPanickingCallbackDoesNotSilenceTheReport(t *testing.T) {
 	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 pause_at=4")
+	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 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..b98f4655e 100644
--- a/internal/connector/review2_test.go
+++ b/internal/connector/review2_test.go
@@ -244,11 +244,15 @@ 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())
 	assert.Equal(t, int32(1), resumes.Load())
 }
diff --git a/internal/connector/seam_contract_test.go b/internal/connector/seam_contract_test.go
index 127e46c1c..c05386784 100644
--- a/internal/connector/seam_contract_test.go
+++ b/internal/connector/seam_contract_test.go
@@ -206,8 +206,9 @@ func TestIntakeGivesTheQueueItsLogger(t *testing.T) {
 
 // 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 has stopped. Each
-// edge is reported there, with the depth and the threshold it crossed.
+// 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)
@@ -231,15 +232,17 @@ func TestIntakeReportsTheBacklogsEdgesOnItsLogger(t *testing.T) {
 	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 is let finish before the
-	// rest, so the resume is decided before the recovery.
+	// 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)
-	for range 2 {
-		_, err := queue.Take(ctx)
-		require.NoError(t, err)
-	}
+	_, 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)
@@ -248,22 +251,28 @@ func TestIntakeReportsTheBacklogsEdgesOnItsLogger(t *testing.T) {
 	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"], "reading the feed is paused")
+	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.Contains(t, lines[1], "depth")
+	assert.EqualValues(t, 1, lines[1]["resume_at"])
 
 	assert.Equal(t, "INFO", lines[2]["level"])
-	assert.Contains(t, lines[2]["msg"], "reading the feed resumed")
+	assert.Contains(t, lines[2]["msg"], "full speed again")
+	assert.EqualValues(t, 1, lines[2]["depth"])
 	assert.EqualValues(t, 3, lines[2]["pause_at"])
-	assert.Contains(t, lines[2], "depth")
+	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.

From d7d55874bcde21bc4f3cb68ca9591e7eb5da7d6f Mon Sep 17 00:00:00 2001
From: Rob Zolkos <rob@37signals.com>
Date: Fri, 23 Oct 2026 16:40:32 -0400
Subject: [PATCH 3/3] Report blocked queue producers accurately and pin
 concurrent pause depth

---
 internal/connector/queue.go              | 11 +++----
 internal/connector/queue_test.go         | 37 +++++++++++++++++++++++-
 internal/connector/review2_test.go       |  1 +
 internal/connector/seam_contract_test.go |  2 +-
 4 files changed, 44 insertions(+), 7 deletions(-)

diff --git a/internal/connector/queue.go b/internal/connector/queue.go
index 5895e55d8..87cff1b20 100644
--- a/internal/connector/queue.go
+++ b/internal/connector/queue.go
@@ -281,9 +281,10 @@ func (q *Queue) decide() {
 	}
 	if !q.paused && q.waiting > 0 {
 		q.paused = true
-		// The held depth: a waiting offer has counted its id but cannot put
-		// it in, so it is not part of what the queue holds.
-		q.pending = append(q.pending, queueEdge{kind: edgePause, fire: q.OnPause, depth: q.depth - q.waiting})
+		// 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
@@ -371,10 +372,10 @@ func (q *Queue) report(log *slog.Logger, edge queueEdge) {
 		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; the next event waits for room, and the feed is read only as fast as admission makes room, so the backlog cannot outgrow memory. "+backlogNextStep,
+		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; the feed is read at full speed again",
+		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)
 	}
 }
diff --git a/internal/connector/queue_test.go b/internal/connector/queue_test.go
index 277d1fca6..fa8617d4a 100644
--- a/internal/connector/queue_test.go
+++ b/internal/connector/queue_test.go
@@ -323,7 +323,10 @@ func TestAPanickingCallbackDoesNotSilenceTheReport(t *testing.T) {
 	require.NoError(t, err)
 	var logs bytes.Buffer
 	queue.SetLogger(slog.New(slog.NewTextHandler(&logs, nil)))
-	queue.OnWarn = func(int) { panic("a warning callback panics") }
+	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))
 
@@ -337,6 +340,38 @@ func TestAPanickingCallbackDoesNotSilenceTheReport(t *testing.T) {
 // 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)
diff --git a/internal/connector/review2_test.go b/internal/connector/review2_test.go
index b98f4655e..1c92598f0 100644
--- a/internal/connector/review2_test.go
+++ b/internal/connector/review2_test.go
@@ -254,6 +254,7 @@ func TestQueuePauseTracksEveryBlockedOffer(t *testing.T) {
 	_, 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 c05386784..1702c161d 100644
--- a/internal/connector/seam_contract_test.go
+++ b/internal/connector/seam_contract_test.go
@@ -263,7 +263,7 @@ func TestIntakeReportsTheBacklogsEdgesOnItsLogger(t *testing.T) {
 	assert.EqualValues(t, 1, lines[1]["resume_at"])
 
 	assert.Equal(t, "INFO", lines[2]["level"])
-	assert.Contains(t, lines[2]["msg"], "full speed again")
+	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"])