Report when the connector's backlog warns, pauses and recovers

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.

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