Bugfix/draining paged queues

Makefile

@@ -7,7 +7,7 @@ COVERAGE_FILE=coverage.out
 # Linker flags to strip debug information
 LDFLAGS=-ldflags="-s -w"
 
-.PHONY: all build clean test coverage release help it-clean fixtures
+.PHONY: all build clean fmt vet test coverage release help it-clean fixtures
 
 # Name of the shared, reused integration-test broker container.
 IT_BROKER=artemisctl-it-broker
@@ -22,12 +22,20 @@ build:
 	go build -o $(BUILD_DIR)/$(BINARY_NAME) $(MAIN_PATH)
 	@echo "==> Done. Binary is in $(BUILD_DIR)/$(BINARY_NAME)"
 
-## test: Run all tests (including our Testcontainers broker tests)
-test:
-	@echo "==> Running linting..."
+## fmt: Format every Go file in the module (gofmt -s, in place)
+fmt:
+	@echo "==> Formatting..."
 	gofmt -s -w .
+	@echo "==> Done."
+
+## vet: Run go vet over the whole module
+vet:
 	@echo "==> Running go vet..."
 	go vet ./...
+	@echo "==> Done."
+
+## test: Run all tests (including our Testcontainers broker tests)
+test: fmt vet
 	@echo "==> Running tests..."
 	go test -v -p 1 ./internal/...
 
@@ -86,7 +94,9 @@ help:
 	@echo ""
 	@echo "Targets:"
 	@echo "  build    - Compile the CLI for your current operating system"
-	@echo "  test     - Run all tests"
+	@echo "  fmt      - Format every Go file (gofmt -s -w)"
+	@echo "  vet      - Run go vet over the whole module"
+	@echo "  test     - Format, vet, then run all tests"
 	@echo "  coverage - Run tests with coverage and generate an HTML report"
 	@echo "  release  - Cross-compile the CLI for Linux, macOS, and Windows"
 	@echo "  fixtures - Regenerate internal/journal/testdata from a live 2.42 container"

internal/broker/drain.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/Azure/go-amqp"
@@ -17,20 +18,184 @@ type RecordSink interface {
 	Sync() error
 }
 
-// DrainQueue consumes every message currently on queue, persisting each
-// batch to sink (Append + Sync/fsync) before acking the broker, so a crash
-// mid-drain never loses a message: it is either still on the broker
-// (unacked) or durably on disk (or both, which is safe to re-drain/replay).
+// PartialDrainError reports that a drain finished with messages still on the
+// broker. It carries the broker's own counters at the moment the drain gave
+// up, so the caller can see WHY the rest did not come: scheduled for later,
+// in flight to another consumer, or a paused queue.
 //
-// Idle detection: each Receive uses a per-call context with timeout idle;
-// a context.DeadlineExceeded from Receive is treated as "queue is empty",
-// not an error, and ends the drain — but only when the outer ctx is still
-// live. If the caller's own ctx has expired/been canceled, that deadline
-// propagates to the child context too, so DeadlineExceeded alone can't
-// distinguish "queue idle" from "caller ran out of time"; we disambiguate
-// against ctx.Err() below so a caller-timeout is reported as an error
-// instead of a false "fully drained".
+// This exists because an idle receive is not proof of an empty queue, and an
+// export that quietly returns a partial file is worse than one that fails.
+type PartialDrainError struct {
+	Queue   string
+	Drained int
+	Stat    QueueStat
+
+	// Scanned reports whether a countMessages scan ran; Counted is what it
+	// found. Stat.MessageCount is only a counter, so when a scan finds 0 while
+	// the counter reports a backlog, the counter has drifted and the
+	// "remaining" messages do not exist -- no drain can ever retrieve them, and
+	// saying so is the difference between a broker bug and a tool bug.
+	//
+	// Scanned gates Counted so that the zero value means "no scan ran" rather
+	// than the far more alarming "a scan found nothing".
+	Scanned bool
+	Counted int64
+}
+
+// outcome is what the broker's counters say about a finished drain pass.
+type outcome int
+
+const (
+	// drainComplete: the broker holds nothing more.
+	drainComplete outcome = iota
+	// drainRetry: messages remain and the broker would hand them over, so the
+	// quiet gap was a stall (GC, depaging, slow network) rather than an end.
+	drainRetry
+	// drainStuck: messages remain that the broker will not deliver to us now.
+	drainStuck
+)
+
+// maxFruitlessPasses bounds how many times in a row DrainQueue will re-attempt
+// a queue that the broker says still holds deliverable messages but that hands
+// over nothing. One fruitless pass is expected and harmless (the broker can
+// still be settling the previous pass's acks, or briefly stalled), but a queue
+// that keeps promising messages it never delivers must fail rather than spin.
+// Each pass already costs a full idle timeout, so this stays cheap.
+const maxFruitlessPasses = 3
+
+func (e *PartialDrainError) Error() string {
+	reasons := make([]string, 0, 3)
+	if e.Stat.ScheduledCount > 0 {
+		reasons = append(reasons, fmt.Sprintf("%d scheduled for later delivery", e.Stat.ScheduledCount))
+	}
+	if e.Stat.DeliveringCount > 0 && e.Stat.ConsumerCount > 0 {
+		reasons = append(reasons, fmt.Sprintf("%d in flight to %d other consumer(s)",
+			e.Stat.DeliveringCount, e.Stat.ConsumerCount))
+	}
+	if e.Stat.Paused {
+		reasons = append(reasons, "queue is paused")
+	}
+	// With no counter to explain the remainder, the broker is telling us these
+	// messages ARE deliverable yet handing over none of them -- a broker-side
+	// stall, not a queue holding messages back. Say exactly that: "the broker
+	// stopped delivering them" reads like a tool giving up, and leaves the
+	// operator with nowhere to look.
+	why := fmt.Sprintf("all %d are deliverable now, but the broker handed over none across %d retries"+
+		"; this is a broker-side stall, not a queue holding them back -- check the broker log and its disk",
+		e.Stat.DeliverableNow(), maxFruitlessPasses)
+	if e.Scanned && e.Counted == 0 && e.Stat.MessageCount > 0 {
+		// The counter advertises a backlog that a scan of the queue cannot
+		// find. Nothing is recoverable here and no retry will help: say so
+		// plainly rather than let it read as an export that failed.
+		why = fmt.Sprintf("messageCount reports %d but a countMessages scan finds 0"+
+			": the queue's message counter has drifted and those messages do not exist"+
+			" -- nothing is left to export; restart the broker to rebuild the counter from its journal",
+			e.Stat.MessageCount)
+	} else if e.Scanned && e.Counted > 0 && len(reasons) == 0 {
+		why = fmt.Sprintf("a scan confirms %d real message(s) and all are deliverable now,"+
+			" yet the broker handed over none across %d retries"+
+			": this is a broker-side stall -- check the broker log and its disk",
+			e.Counted, maxFruitlessPasses)
+	}
+	if len(reasons) > 0 {
+		why = strings.Join(reasons, ", ")
+	}
+	return fmt.Sprintf("incomplete drain of %s: took %d message(s), but %d remain on the broker (%s)",
+		e.Queue, e.Drained, e.Stat.MessageCount, why)
+}
+
+// DrainQueue consumes every message on queue, persisting each batch to sink
+// (Append + Sync/fsync) before acking the broker, so a crash mid-drain never
+// loses a message: it is either still on the broker (unacked) or durably on
+// disk (or both, which is safe to re-drain/replay).
+//
+// Completion is decided by the broker, not by silence. A gap in delivery only
+// ends a drain *pass*; DrainQueue then asks the broker for the queue's real
+// depth and acts on it:
+//
+//   - depth 0: the queue is genuinely empty, the drain is complete.
+//   - messages remain and are deliverable: the gap was a stall (broker GC,
+//     depaging, a slow network), so it drains again.
+//   - messages remain but the broker is holding them back (scheduled, in
+//     flight to another consumer, paused queue): no amount of waiting will
+//     produce them, so it returns *PartialDrainError with the counters.
+//
+// The earlier behaviour — treat any idle gap as "queue drained" and return nil
+// — silently under-drained: a queue reporting 210K messages could export ~50K
+// and still exit 0, because most of the depth was never deliverable to us.
 func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink, idle time.Duration, batch int) (int, error) {
+	total := 0
+	fruitless := 0
+	for {
+		n, err := c.drainPass(ctx, queue, sink, idle, batch)
+		total += n
+		if err != nil {
+			return total, err
+		}
+		if n > 0 {
+			fruitless = 0
+		} else {
+			fruitless++
+		}
+
+		stat, err := c.QueueStatByName(ctx, queue)
+		if err != nil {
+			if errors.Is(err, ErrQueueNotFound) {
+				// Auto-delete removed the queue once we emptied it; nothing
+				// is left behind, so the drain is complete.
+				return total, nil
+			}
+			return total, fmt.Errorf("verify drain of %s: %w", queue, err)
+		}
+		switch drainOutcome(stat, fruitless) {
+		case drainComplete:
+			return total, nil
+		case drainRetry:
+			continue
+		default:
+			// Only now, on the failure path, pay for a scan: countMessages
+			// walks the queue, so it is far more expensive than the counter
+			// read above and is worth it exactly once, to say WHY.
+			pde := &PartialDrainError{Queue: queue, Drained: total, Stat: stat}
+			if n, cErr := c.CountMessages(ctx, queue); cErr == nil {
+				pde.Scanned, pde.Counted = true, n
+			}
+			return total, pde
+		}
+	}
+}
+
+// drainOutcome decides what a pass means, given the broker's counters
+// afterwards and how many consecutive passes have now come back empty
+// (fruitless == 0 means the last pass took messages). It is the whole of the
+// "is this drain actually finished?" judgement, kept pure so every case is
+// testable without a broker.
+func drainOutcome(stat QueueStat, fruitless int) outcome {
+	if stat.MessageCount == 0 {
+		return drainComplete
+	}
+	// Retrying is only worth it if the broker would actually hand the
+	// remainder over -- otherwise we would spin forever on messages we cannot
+	// have -- and only while passes are still producing something.
+	if stat.DeliverableNow() > 0 && fruitless < maxFruitlessPasses {
+		return drainRetry
+	}
+	return drainStuck
+}
+
+// drainPass consumes messages from queue until the broker goes quiet for idle,
+// returning how many it took. A quiet broker ends the pass; it does NOT mean
+// the queue is empty -- only DrainQueue's check against the broker's counters
+// can establish that.
+//
+// Idle detection: each Receive uses a per-call context with timeout idle;
+// a context.DeadlineExceeded from Receive ends the pass, but only when the
+// outer ctx is still live. If the caller's own ctx has expired/been canceled,
+// that deadline propagates to the child context too, so DeadlineExceeded alone
+// can't distinguish "broker quiet" from "caller ran out of time"; we
+// disambiguate against ctx.Err() below so a caller-timeout is reported as an
+// error instead of a false "fully drained".
+func (c *Client) drainPass(ctx context.Context, queue string, sink RecordSink, idle time.Duration, batch int) (int, error) {
 	if batch <= 0 {
 		// Credit: int32(batch) with batch<=0 grants zero link credit, which
 		// silently receives nothing (or reports (0,nil) on a non-empty
@@ -81,7 +246,7 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 					// surface it as an error instead.
 					return total, ctx.Err()
 				}
-				break // queue idle => drained
+				break // broker quiet => end of pass (NOT proof of an empty queue)
 			}
 			// Messages already Appended-but-not-yet-Acked when this error
 			// aborts the drain stay safely on the broker (no loss), but may
@@ -141,21 +306,33 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 // DrainAll enumerates every queue via ListQueues and drains each one in
 // turn, invoking onQueue (if non-nil) after each queue completes with the
 // queue's name and the number of messages drained from it.
+//
+// A queue that cannot be fully drained does not abort the run: whatever the
+// other queues hold is still worth exporting, so DrainAll records the
+// *PartialDrainError, moves on, and returns every one of them joined together
+// at the end. The messages it did take are already persisted, and the returned
+// count reflects them, but the error means the export is NOT complete.
+// Any other error (a broken connection, a failing sink) aborts immediately.
 func (c *Client) DrainAll(ctx context.Context, sink RecordSink, idle time.Duration, batch int, onQueue func(name string, n int)) (int, error) {
 	queues, err := c.ListQueues(ctx)
 	if err != nil {
 		return 0, err
 	}
 	total := 0
+	var partial []error
 	for _, q := range queues {
 		n, err := c.DrainQueue(ctx, q.Name, sink, idle, batch)
+		total += n
 		if err != nil {
-			return total, err
+			var pde *PartialDrainError
+			if !errors.As(err, &pde) {
+				return total, err
+			}
+			partial = append(partial, err)
 		}
 		if onQueue != nil {
 			onQueue(q.Name, n)
 		}
-		total += n
 	}
-	return total, nil
+	return total, errors.Join(partial...)
 }