From f49c0d1df8e83906232a422c7a3d9e29ce62d973 Mon Sep 17 00:00:00 2001
From: Richard Martikan <ric.martikan@gmail.com>
Date: Wed, 15 Jul 2026 20:12:00 +0200
Subject: [PATCH 1/4] fix: correction for draining queues.

It could happen, that the broker paged the messages, and during that
time when it started to read the paged messages.. the draining was
stopped, because it thought that the queue is already empty.
the draining process had to be hardened, so the whole queue will be
drained doesn't matter how big is it and paged or not.
---
 internal/broker/drain.go                      | 172 ++++++++++--
 .../broker/drain_verify_integration_test.go   | 252 ++++++++++++++++++
 internal/broker/drain_verify_test.go          | 174 ++++++++++++
 internal/broker/management.go                 | 119 ++++++++-
 .../broker/management_integration_test.go     |   9 +-
 internal/cli/export.go                        |  28 +-
 internal/cli/export_integration_test.go       |  99 +++++++
 internal/cli/salvage_integration_test.go      |  14 +-
 8 files changed, 832 insertions(+), 35 deletions(-)
 create mode 100644 internal/broker/drain_verify_integration_test.go
 create mode 100644 internal/broker/drain_verify_test.go

diff --git a/internal/broker/drain.go b/internal/broker/drain.go
index f97fc41..a2be15f 100644
--- a/internal/broker/drain.go
+++ b/internal/broker/drain.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/Azure/go-amqp"
@@ -17,20 +18,147 @@ 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
+}
+
+
+// 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")
+	}
+	why := "the broker stopped delivering them"
+	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:
+			return total, &PartialDrainError{Queue: queue, Drained: total, Stat: stat}
+		}
+	}
+}
+
+
+// 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 +209,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 +269,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...)
 }
diff --git a/internal/broker/drain_verify_integration_test.go b/internal/broker/drain_verify_integration_test.go
new file mode 100644
index 0000000..b5505ed
--- /dev/null
+++ b/internal/broker/drain_verify_integration_test.go
@@ -0,0 +1,252 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/Azure/go-amqp"
+)
+
+// scheduleAt marks every message for delivery at t. Artemis honours the AMQP
+// x-opt-delivery-time annotation for scheduled delivery.
+func scheduleAt(msgs []*amqp.Message, at time.Time) {
+	for _, m := range msgs {
+		m.Annotations = amqp.Annotations{"x-opt-delivery-time": at.UnixMilli()}
+	}
+}
+
+// TestDrainQueueScheduledRemainderIsReported pins the reported bug: a queue
+// whose depth is mostly messages the broker will not deliver yet must NOT look
+// like a completed drain. Before the fix, DrainQueue took the deliverable
+// messages, saw the broker go quiet, and returned nil -- so a DLQ reporting
+// 210K exported ~50K and still exited 0.
+func TestDrainQueueScheduledRemainderIsReported(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("sched-remainder")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(400, 256, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce deliverable: %v", err)
+	}
+	later := GenerateMessages(600, 256, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 4, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	sink := &sliceSink{}
+	n, err := c.DrainQueue(ctx, queue, sink, 3*time.Second, 100)
+
+	var pde *PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("drain returned n=%d err=%v, want *PartialDrainError", n, err)
+	}
+	if n != 400 || len(sink.recs) != 400 {
+		t.Fatalf("drained %d (sink %d), want the 400 deliverable messages", n, len(sink.recs))
+	}
+	if pde.Stat.ScheduledCount != 600 {
+		t.Fatalf("reported ScheduledCount = %d, want 600", pde.Stat.ScheduledCount)
+	}
+	if pde.Stat.MessageCount != 600 {
+		t.Fatalf("reported MessageCount = %d, want 600 left on broker", pde.Stat.MessageCount)
+	}
+	t.Logf("drain correctly refused to claim success: %v", err)
+}
+
+// TestDrainQueuePausedIsReported covers the other way a queue can be quiet but
+// non-empty: a paused queue dispatches nothing at all, so every message stays.
+func TestDrainQueuePausedIsReported(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("paused-remainder")
+	if _, err := c.Produce(ctx, queue, GenerateMessages(25, 256, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+	if _, err := c.callManagement(ctx, "queue."+queue, "pause", "[]"); err != nil {
+		t.Fatalf("pause queue: %v", err)
+	}
+	defer func() {
+		_, _ = c.callManagement(context.Background(), "queue."+queue, "resume", "[]")
+		_, _ = c.PurgeQueue(context.Background(), queue)
+	}()
+
+	n, err := c.DrainQueue(ctx, queue, &sliceSink{}, 2*time.Second, 100)
+
+	var pde *PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("drain returned n=%d err=%v, want *PartialDrainError", n, err)
+	}
+	if !pde.Stat.Paused {
+		t.Fatalf("reported Stat.Paused = false, want true; stat=%+v", pde.Stat)
+	}
+	if pde.Stat.MessageCount != 25 {
+		t.Fatalf("reported MessageCount = %d, want 25", pde.Stat.MessageCount)
+	}
+}
+
+// TestDrainQueueFullyDrainsWhenTrulyEmpty guards the other direction: the new
+// broker-verified completion check must not turn an ordinary complete drain
+// into an error.
+func TestDrainQueueFullyDrainsWhenTrulyEmpty(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("fully-drains")
+	if _, err := c.Produce(ctx, queue, GenerateMessages(750, 256, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	sink := &sliceSink{}
+	n, err := c.DrainQueue(ctx, queue, sink, 2*time.Second, 100)
+	if err != nil {
+		t.Fatalf("drain: %v", err)
+	}
+	if n != 750 || len(sink.recs) != 750 {
+		t.Fatalf("drained %d (sink %d), want 750", n, len(sink.recs))
+	}
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil && !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("stat: %v", err)
+	}
+	if err == nil && stat.MessageCount != 0 {
+		t.Fatalf("broker still holds %d messages after a nil-error drain", stat.MessageCount)
+	}
+}
+
+// TestQueueStatByNameReportsCounters pins the decoding of the counters the
+// drain's completion check relies on, against a real broker reply.
+func TestQueueStatByNameReportsCounters(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("stat-counters")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(10, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce now: %v", err)
+	}
+	later := GenerateMessages(15, 128, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil {
+		t.Fatalf("QueueStatByName: %v", err)
+	}
+	if stat.Name != queue || stat.MessageCount != 25 || stat.ScheduledCount != 15 {
+		t.Fatalf("stat = %+v, want name=%s messageCount=25 scheduledCount=15", stat, queue)
+	}
+	if stat.Paused {
+		t.Fatalf("stat.Paused = true on a live queue: %+v", stat)
+	}
+	if got := stat.DeliverableNow(); got != 10 {
+		t.Fatalf("DeliverableNow() = %d, want 10", got)
+	}
+}
+
+// TestQueueStatByNameMissingQueue pins the not-found path, which DrainQueue
+// treats as "the queue was auto-deleted once emptied", not as a failure.
+func TestQueueStatByNameMissingQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	if _, err := c.QueueStatByName(ctx, "no-such-queue-9f3a"); !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("err = %v, want ErrQueueNotFound", err)
+	}
+}
+
+// TestPurgeQueueRemovesScheduled pins that purge clears what a drain cannot,
+// which is what resetBroker relies on to give each test a clean slate.
+func TestPurgeQueueRemovesScheduled(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("purge-scheduled")
+	msgs := GenerateMessages(30, 128, nil)
+	scheduleAt(msgs, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, msgs, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	removed, err := c.PurgeQueue(ctx, queue)
+	if err != nil {
+		t.Fatalf("purge: %v", err)
+	}
+	if removed != 30 {
+		t.Fatalf("purge removed %d, want 30", removed)
+	}
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil && !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("stat: %v", err)
+	}
+	if err == nil && stat.MessageCount != 0 {
+		t.Fatalf("queue still holds %d after purge", stat.MessageCount)
+	}
+}
diff --git a/internal/broker/drain_verify_test.go b/internal/broker/drain_verify_test.go
new file mode 100644
index 0000000..a92385f
--- /dev/null
+++ b/internal/broker/drain_verify_test.go
@@ -0,0 +1,174 @@
+package broker
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestDeliverableNow(t *testing.T) {
+	tests := []struct {
+		name string
+		stat QueueStat
+		want int64
+	}{
+		{"empty queue", QueueStat{}, 0},
+		{"all deliverable", QueueStat{MessageCount: 10}, 10},
+		{"scheduled held back", QueueStat{MessageCount: 10, ScheduledCount: 6}, 4},
+		{"in flight to another consumer", QueueStat{MessageCount: 10, DeliveringCount: 3, ConsumerCount: 1}, 7},
+		// No consumer holds these: the broker is returning them to the queue.
+		{"in flight with no consumer is settlement lag", QueueStat{MessageCount: 10, DeliveringCount: 10}, 10},
+		{"scheduled and delivering", QueueStat{MessageCount: 10, ScheduledCount: 6, DeliveringCount: 3, ConsumerCount: 1}, 1},
+		{"wholly scheduled", QueueStat{MessageCount: 10, ScheduledCount: 10}, 0},
+		{"paused ignores counters", QueueStat{MessageCount: 10, Paused: true}, 0},
+		// The counters are sampled independently and can overlap in flight.
+		{"overlapping counters never go negative", QueueStat{MessageCount: 5, ScheduledCount: 4, DeliveringCount: 4, ConsumerCount: 1}, 0},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.stat.DeliverableNow(); got != tc.want {
+				t.Fatalf("DeliverableNow() = %d, want %d", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestDrainOutcome(t *testing.T) {
+	tests := []struct {
+		name      string
+		stat      QueueStat
+		fruitless int
+		want      outcome
+	}{
+		{
+			name: "empty queue is complete",
+			stat: QueueStat{MessageCount: 0},
+			want: drainComplete,
+		},
+		{
+			name:      "empty queue is complete even after fruitless passes",
+			stat:      QueueStat{MessageCount: 0},
+			fruitless: maxFruitlessPasses,
+			want:      drainComplete,
+		},
+		{
+			name: "deliverable remainder after progress means the gap was a stall",
+			stat: QueueStat{MessageCount: 500},
+			want: drainRetry,
+		},
+		{
+			name:      "one fruitless pass is tolerated: acks may still be settling",
+			stat:      QueueStat{MessageCount: 500},
+			fruitless: 1,
+			want:      drainRetry,
+		},
+		{
+			name:      "a queue that keeps promising messages it never delivers gives up",
+			stat:      QueueStat{MessageCount: 500},
+			fruitless: maxFruitlessPasses,
+			want:      drainStuck,
+		},
+		{
+			name: "in-flight remainder with no consumer is retried, not reported stuck",
+			stat: QueueStat{MessageCount: 50, DeliveringCount: 50},
+			want: drainRetry,
+		},
+		{
+			name: "scheduled remainder is stuck, retrying cannot help",
+			stat: QueueStat{MessageCount: 600, ScheduledCount: 600},
+			want: drainStuck,
+		},
+		{
+			name: "remainder in flight to another consumer is stuck",
+			stat: QueueStat{MessageCount: 50, DeliveringCount: 50, ConsumerCount: 2},
+			want: drainStuck,
+		},
+		{
+			name: "paused queue is stuck",
+			stat: QueueStat{MessageCount: 50, Paused: true},
+			want: drainStuck,
+		},
+		{
+			name: "partly scheduled remainder still has deliverable messages to take",
+			stat: QueueStat{MessageCount: 600, ScheduledCount: 400},
+			want: drainRetry,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := drainOutcome(tc.stat, tc.fruitless); got != tc.want {
+				t.Fatalf("drainOutcome() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestPartialDrainErrorMessage(t *testing.T) {
+	tests := []struct {
+		name  string
+		err   *PartialDrainError
+		wants []string
+	}{
+		{
+			name: "scheduled",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 50701,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 160000, ScheduledCount: 160000},
+			},
+			wants: []string{"incomplete drain of DLQ", "50701", "160000", "scheduled for later"},
+		},
+		{
+			name: "in flight to another consumer",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 5,
+				Stat:    QueueStat{Name: "orders", MessageCount: 20, DeliveringCount: 20, ConsumerCount: 3},
+			},
+			wants: []string{"incomplete drain of orders", "in flight to 3 other consumer(s)"},
+		},
+		{
+			name: "paused",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 0,
+				Stat:    QueueStat{Name: "orders", MessageCount: 7, Paused: true},
+			},
+			wants: []string{"queue is paused"},
+		},
+		{
+			name: "deliverable but the broker went quiet: no counter explains it",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 3,
+				Stat:    QueueStat{Name: "orders", MessageCount: 9},
+			},
+			wants: []string{"the broker stopped delivering them"},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := tc.err.Error()
+			for _, want := range tc.wants {
+				if !strings.Contains(got, want) {
+					t.Fatalf("Error() = %q, want it to mention %q", got, want)
+				}
+			}
+		})
+	}
+}
+
+// A PartialDrainError must survive errors.Join + errors.As, which is how
+// DrainAll reports several bad queues at once and how export detects one.
+func TestPartialDrainErrorIsUnwrappable(t *testing.T) {
+	pde := &PartialDrainError{Queue: "DLQ", Drained: 1, Stat: QueueStat{MessageCount: 2}}
+	joined := errors.Join(errors.New("other queue failed"), pde)
+
+	var got *PartialDrainError
+	if !errors.As(joined, &got) {
+		t.Fatal("errors.As did not find *PartialDrainError in a joined error")
+	}
+	if got.Queue != "DLQ" {
+		t.Fatalf("unwrapped Queue = %q, want DLQ", got.Queue)
+	}
+}
diff --git a/internal/broker/management.go b/internal/broker/management.go
index e36157d..603e996 100644
--- a/internal/broker/management.go
+++ b/internal/broker/management.go
@@ -9,12 +9,50 @@ import (
 	"github.com/Azure/go-amqp"
 )
 
-// QueueStat is one queue's name and current message depth, as reported by the
-// broker's listQueues management operation. MessageCount arrives as a
-// JSON string and is decoded into an int64.
+// QueueStat is one queue's depth and delivery state, as reported by the
+// broker's listQueues management operation. Every counter arrives as a JSON
+// string and is decoded into its Go type.
+//
+// MessageCount is the total depth, and it is NOT the number of messages a
+// consumer can receive right now: it also counts messages the broker will not
+// hand out yet. ScheduledCount (scheduled delivery / redelivery delay, not due
+// yet) and DeliveringCount (already dispatched to some consumer, awaiting
+// settlement) are both included in it, and a Paused queue dispatches nothing at
+// all regardless of the counters. DeliverableNow is the subset a drain can
+// actually expect to receive.
 type QueueStat struct {
-	Name         string `json:"name"`
-	MessageCount int64  `json:"messageCount,string"`
+	Name            string `json:"name"`
+	MessageCount    int64  `json:"messageCount,string"`
+	ScheduledCount  int64  `json:"scheduledCount,string"`
+	DeliveringCount int64  `json:"deliveringCount,string"`
+	ConsumerCount   int64  `json:"consumerCount,string"`
+	Paused          bool   `json:"paused,string"`
+}
+
+// DeliverableNow reports how many of the queue's messages the broker would
+// hand to a consumer right now: the depth minus the messages it is holding
+// back. A paused queue delivers nothing, so it always reports 0.
+//
+// DeliveringCount only counts against the total while some consumer is
+// attached to hold those messages. With no consumers, an in-flight count is
+// just settlement lag -- typically our own just-closed receiver, whose unacked
+// messages the broker is in the middle of returning to the queue -- and those
+// messages ARE coming back, so they must not be mistaken for messages we can
+// never have.
+func (q QueueStat) DeliverableNow() int64 {
+	if q.Paused {
+		return 0
+	}
+	n := q.MessageCount - q.ScheduledCount
+	if q.ConsumerCount > 0 {
+		n -= q.DeliveringCount
+	}
+	if n < 0 {
+		// The counters are sampled independently and can overlap in flight;
+		// never report a negative backlog.
+		return 0
+	}
+	return n
 }
 
 // callManagement performs a request/reply against activemq.management.
@@ -67,7 +105,76 @@ func (c *Client) ListQueues(ctx context.Context) ([]QueueStat, error) {
 	return parseQueueStatsReply(reply)
 }
 
+// ErrQueueNotFound is returned by QueueStatByName when the broker does not
+// list the named queue, which normally means it has been auto-deleted.
+var ErrQueueNotFound = fmt.Errorf("queue not found on broker")
+
+// QueueStatByName returns the broker's current stats for a single queue. It
+// asks the broker to filter by exact name rather than listing every queue, so
+// it stays cheap enough to call repeatedly during a drain. Unlike ListQueues it
+// does not filter internal queues: the caller already named the queue it wants.
+func (c *Client) QueueStatByName(ctx context.Context, name string) (QueueStat, error) {
+	filter, err := json.Marshal(map[string]string{
+		"field": "name", "operation": "EQUALS", "value": name,
+		"sortField": "messageCount", "sortOrder": "desc",
+	})
+	if err != nil {
+		return QueueStat{}, fmt.Errorf("marshal queue filter: %w", err)
+	}
+	body, err := json.Marshal([]interface{}{string(filter), 1, 1})
+	if err != nil {
+		return QueueStat{}, fmt.Errorf("marshal listQueues args: %w", err)
+	}
+	reply, err := c.callManagement(ctx, "broker", "listQueues", string(body))
+	if err != nil {
+		return QueueStat{}, err
+	}
+	stats, err := parseQueueStatsRaw(reply)
+	if err != nil {
+		return QueueStat{}, err
+	}
+	for _, q := range stats {
+		if q.Name == name {
+			return q, nil
+		}
+	}
+	return QueueStat{}, fmt.Errorf("%w: %s", ErrQueueNotFound, name)
+}
+
+// PurgeQueue removes every message from a queue, including messages a consumer
+// cannot receive (scheduled, or held for redelivery). It returns the number of
+// messages removed. This is destructive and does not persist anything: it is
+// for resetting a queue, not for exporting it.
+func (c *Client) PurgeQueue(ctx context.Context, name string) (int64, error) {
+	reply, err := c.callManagement(ctx, "queue."+name, "removeAllMessages", "[]")
+	if err != nil {
+		return 0, fmt.Errorf("purge %s: %w", name, err)
+	}
+	// The reply value is the removed count, encoded as a JSON array holding a
+	// single number ("[12]"), matching the other management replies' shape.
+	raw, ok := reply.Value.(string)
+	if !ok {
+		return 0, nil
+	}
+	var outer []int64
+	if err := json.Unmarshal([]byte(raw), &outer); err != nil || len(outer) == 0 {
+		return 0, nil
+	}
+	return outer[0], nil
+}
+
 func parseQueueStatsReply(reply *amqp.Message) ([]QueueStat, error) {
+	stats, err := parseQueueStatsRaw(reply)
+	if err != nil {
+		return nil, err
+	}
+	return filterInternalQueues(stats), nil
+}
+
+// parseQueueStatsRaw decodes a listQueues reply without filtering internal
+// queues, so a caller asking for one queue by name gets it back whatever it is
+// called.
+func parseQueueStatsRaw(reply *amqp.Message) ([]QueueStat, error) {
 	strVal, ok := reply.Value.(string)
 	if !ok {
 		return nil, fmt.Errorf("unexpected management response format")
@@ -85,7 +192,7 @@ func parseQueueStatsReply(reply *amqp.Message) ([]QueueStat, error) {
 	if err := json.Unmarshal([]byte(outer[0]), &paged); err != nil {
 		return nil, fmt.Errorf("parse paged response: %w", err)
 	}
-	return filterInternalQueues(paged.Data), nil
+	return paged.Data, nil
 }
 
 func filterInternalQueues(in []QueueStat) []QueueStat {
diff --git a/internal/broker/management_integration_test.go b/internal/broker/management_integration_test.go
index 0f493ae..44fe6c6 100644
--- a/internal/broker/management_integration_test.go
+++ b/internal/broker/management_integration_test.go
@@ -77,8 +77,13 @@ func resetBroker(t testing.TB, props ConnectionProps) {
 		t.Fatalf("reset list queues: %v", err)
 	}
 	for _, q := range qs {
-		if _, err := c.DrainQueue(ctx, q.Name, discardSink{}, 500*time.Millisecond, 200); err != nil {
-			t.Fatalf("reset drain %s: %v", q.Name, err)
+		// Purge rather than drain: removeAllMessages also clears messages a
+		// consumer cannot receive (scheduled, held for redelivery), which a
+		// drain leaves behind -- and which DrainQueue now correctly reports as
+		// an incomplete drain, so a single leftover scheduled message from an
+		// earlier test would fail every later one.
+		if _, err := c.PurgeQueue(ctx, q.Name); err != nil {
+			t.Fatalf("reset purge %s: %v", q.Name, err)
 		}
 	}
 }
diff --git a/internal/cli/export.go b/internal/cli/export.go
index c3d848a..1a85549 100644
--- a/internal/cli/export.go
+++ b/internal/cli/export.go
@@ -48,22 +48,32 @@ func newExportCmd() *cobra.Command {
 				return fmt.Errorf("remove stale checkpoint: %w", err)
 			}
 
-			total, err := c.DrainAll(ctx, w, drainTimeout, batch, func(q string, n int) {
+			total, drainErr := c.DrainAll(ctx, w, drainTimeout, batch, func(q string, n int) {
 				fmt.Fprintf(cmd.OutOrStdout(), "drained %d from %s\n", n, q)
 			})
-			if err != nil {
+			// Whatever was drained is worth keeping even when the drain did not
+			// complete, so fsync before reporting either way.
+			syncErr := w.Sync()
+			if drainErr == nil && syncErr != nil {
+				return syncErr
+			}
+			if drainErr != nil {
 				// SIGINT: drained records are already fsync'd; report a clean
 				// interruption rather than a crash. Re-running export needs a
 				// fresh --out path (an existing non-empty store is refused).
-				if errors.Is(err, context.Canceled) {
-					_ = w.Sync()
+				if errors.Is(drainErr, context.Canceled) {
 					fmt.Fprintf(cmd.OutOrStdout(), "interrupted after %d messages, progress saved to %s\n", total, out)
-					return err
+					return drainErr
 				}
-				return err
-			}
-			if err := w.Sync(); err != nil {
-				return err
+				var pde *broker.PartialDrainError
+				if errors.As(drainErr, &pde) {
+					// The store holds real, replayable messages -- it is just
+					// not the whole queue. Say so on stdout so the count is not
+					// mistaken for a complete export, and still fail: exporting
+					// a subset while exiting 0 is how messages get lost.
+					fmt.Fprintf(cmd.OutOrStdout(), "INCOMPLETE: saved %d messages to %s, but the broker still holds messages\n", total, out)
+				}
+				return drainErr
 			}
 			fmt.Fprintf(cmd.OutOrStdout(), "exported %d messages to %s\n", total, out)
 			return nil
diff --git a/internal/cli/export_integration_test.go b/internal/cli/export_integration_test.go
index 313a977..97aa103 100644
--- a/internal/cli/export_integration_test.go
+++ b/internal/cli/export_integration_test.go
@@ -2,9 +2,18 @@
 package cli
 
 import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
+
+	"github.com/Azure/go-amqp"
+	"github.com/martikan/artemisctl/internal/broker"
 )
 
 func TestExportCommandCreatesStore(t *testing.T) {
@@ -30,3 +39,93 @@ func TestExportCommandCreatesStore(t *testing.T) {
 		t.Fatalf("store not written: %v size=%v", err, fi)
 	}
 }
+
+// TestExportFailsOnPartialDrain pins the export contract that the reported bug
+// broke: when the broker still holds messages the drain could not take, export
+// must NOT report success. Previously it printed "exported N messages" and
+// exited 0 while most of a 210K-deep DLQ stayed on the broker, which is how a
+// partial store gets mistaken for a full one.
+//
+// The undrainable remainder here is a scheduled message: counted in the
+// queue's depth, but not deliverable to any consumer until its delivery time.
+func TestExportFailsOnPartialDrain(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemisForCLI(t)
+	queue := fmt.Sprintf("export.partial.%d", time.Now().UnixNano())
+
+	seedQueue(t, props, queue, []string{"take-me", "and-me"})
+	seedScheduled(t, props, queue, 3, time.Now().Add(time.Hour))
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	c, err := broker.Connect(ctx, props)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer c.Close(context.Background())
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	out := filepath.Join(t.TempDir(), "partial.artx")
+	var stdout bytes.Buffer
+	root := NewRootCmd()
+	root.SetOut(&stdout)
+	root.SetArgs([]string{
+		"export", "--out", out,
+		"--url", props.URL, "-u", props.Username, "-p", props.Password,
+		"--drain-timeout", "2s",
+	})
+
+	err = root.Execute()
+	if err == nil {
+		t.Fatalf("export reported success on a partial drain; stdout:\n%s", stdout.String())
+	}
+	var pde *broker.PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("export err = %v, want *PartialDrainError", err)
+	}
+	if got := stdout.String(); !strings.Contains(got, "INCOMPLETE") {
+		t.Fatalf("export stdout must flag the partial store; got:\n%s", got)
+	}
+	if strings.Contains(stdout.String(), "exported ") {
+		t.Fatalf("export printed a success headline on a partial drain:\n%s", stdout.String())
+	}
+	// The messages it DID drain must still be durably saved: a failed export is
+	// not a reason to throw away recovered messages.
+	fi, statErr := os.Stat(out)
+	if statErr != nil || fi.Size() <= 5 {
+		t.Fatalf("partial store not written: %v size=%v", statErr, fi)
+	}
+}
+
+// seedScheduled puts n messages on queue that the broker will hold until at,
+// so they count towards the queue's depth but cannot be drained.
+func seedScheduled(t *testing.T, props broker.ConnectionProps, queue string, n int, at time.Time) {
+	t.Helper()
+	ctx := context.Background()
+	conn, err := amqp.Dial(ctx, "amqp://"+props.URL, &amqp.ConnOptions{
+		SASLType: amqp.SASLTypePlain(props.Username, props.Password),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	sess, err := conn.NewSession(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sender, err := sess.NewSender(ctx, queue, &amqp.SenderOptions{TargetCapabilities: []string{"queue"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sender.Close(ctx)
+	for i := 0; i < n; i++ {
+		msg := amqp.NewMessage([]byte(fmt.Sprintf("scheduled-%d", i)))
+		msg.Properties = &amqp.MessageProperties{MessageID: fmt.Sprintf("sched-%d-%d", time.Now().UnixNano(), i)}
+		msg.Annotations = amqp.Annotations{"x-opt-delivery-time": at.UnixMilli()}
+		if err := sender.Send(ctx, msg, nil); err != nil {
+			t.Fatalf("send scheduled: %v", err)
+		}
+	}
+}
diff --git a/internal/cli/salvage_integration_test.go b/internal/cli/salvage_integration_test.go
index aa8b6ce..7fbe97a 100644
--- a/internal/cli/salvage_integration_test.go
+++ b/internal/cli/salvage_integration_test.go
@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -266,8 +267,17 @@ func TestSalvageE2E(t *testing.T) {
 	dctx, dcancel := context.WithTimeout(context.Background(), 60*time.Second)
 	n, err := c.DrainQueue(dctx, redirectQueue, sink, 1*time.Second, 200)
 	dcancel()
-	if err != nil {
-		t.Fatalf("DrainQueue: %v", err)
+	// The scheduled record stays on the queue, so this drain is by definition
+	// incomplete and DrainQueue must say so rather than return nil: a drain
+	// that leaves messages behind while reporting success is what let an export
+	// silently ship a partial store. The rest of the queue must still have been
+	// drained in full.
+	var pde *broker.PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("DrainQueue err = %v, want *PartialDrainError for the undrainable scheduled record", err)
+	}
+	if pde.Stat.MessageCount != 1 || pde.Stat.ScheduledCount != 1 {
+		t.Fatalf("DrainQueue reported %+v, want exactly 1 remaining, scheduled", pde.Stat)
 	}
 	if n != wantDrainable {
 		t.Fatalf("DrainQueue drained %d messages, want %d (%d total minus the not-yet-due scheduled record, which cannot be drained until 2100-01-01)",

From 2fe7a77366c52da4215565273e355867f02c5305 Mon Sep 17 00:00:00 2001
From: Richard Martikan <ric.martikan@gmail.com>
Date: Thu, 16 Jul 2026 20:03:55 +0200
Subject: [PATCH 2/4] Add queue drift detection

---
 internal/broker/drain.go                      |  43 ++++-
 .../broker/drain_verify_integration_test.go   |  50 ++++++
 internal/broker/drain_verify_test.go          |  57 ++++++-
 internal/broker/drift.go                      | 152 ++++++++++++++++++
 internal/broker/drift_integration_test.go     | 126 +++++++++++++++
 internal/broker/drift_test.go                 | 113 +++++++++++++
 internal/broker/management.go                 |  53 +++++-
 internal/broker/management_test.go            |  18 ++-
 internal/cli/commands_integration_test.go     |  29 ++++
 internal/cli/status.go                        |  59 ++++++-
 10 files changed, 685 insertions(+), 15 deletions(-)
 create mode 100644 internal/broker/drift.go
 create mode 100644 internal/broker/drift_integration_test.go
 create mode 100644 internal/broker/drift_test.go

diff --git a/internal/broker/drain.go b/internal/broker/drain.go
index a2be15f..0f71954 100644
--- a/internal/broker/drain.go
+++ b/internal/broker/drain.go
@@ -29,6 +29,17 @@ 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
 }
 
 
@@ -65,7 +76,28 @@ func (e *PartialDrainError) Error() string {
 	if e.Stat.Paused {
 		reasons = append(reasons, "queue is paused")
 	}
-	why := "the broker stopped delivering them"
+	// 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, ", ")
 	}
@@ -122,7 +154,14 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 		case drainRetry:
 			continue
 		default:
-			return total, &PartialDrainError{Queue: queue, Drained: total, Stat: stat}
+			// 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
 		}
 	}
 }
diff --git a/internal/broker/drain_verify_integration_test.go b/internal/broker/drain_verify_integration_test.go
index b5505ed..79d7682 100644
--- a/internal/broker/drain_verify_integration_test.go
+++ b/internal/broker/drain_verify_integration_test.go
@@ -191,6 +191,56 @@ func TestQueueStatByNameReportsCounters(t *testing.T) {
 	}
 }
 
+// TestCountMessagesScansTheQueue pins CountMessages against a real broker.
+// Unlike QueueStat.MessageCount it walks the queue rather than reading a
+// counter, which is what lets a stalled drain tell a real backlog apart from a
+// drifted counter advertising messages that do not exist.
+func TestCountMessagesScansTheQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("count-scan")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(12, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+	// Scheduled messages are still messages on the queue: the scan counts them
+	// even though a consumer cannot receive them yet.
+	later := GenerateMessages(3, 128, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	got, err := c.CountMessages(ctx, queue)
+	if err != nil {
+		t.Fatalf("CountMessages: %v", err)
+	}
+	if got != 15 {
+		t.Fatalf("CountMessages = %d, want 15", got)
+	}
+	// The whole point is that the scan agrees with the counter on a healthy
+	// queue; only then does a disagreement mean drift.
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil {
+		t.Fatalf("QueueStatByName: %v", err)
+	}
+	if stat.MessageCount != got {
+		t.Fatalf("scan found %d but messageCount reports %d on a healthy queue", got, stat.MessageCount)
+	}
+}
+
 // TestQueueStatByNameMissingQueue pins the not-found path, which DrainQueue
 // treats as "the queue was auto-deleted once emptied", not as a failure.
 func TestQueueStatByNameMissingQueue(t *testing.T) {
diff --git a/internal/broker/drain_verify_test.go b/internal/broker/drain_verify_test.go
index a92385f..8465714 100644
--- a/internal/broker/drain_verify_test.go
+++ b/internal/broker/drain_verify_test.go
@@ -143,7 +143,62 @@ func TestPartialDrainErrorMessage(t *testing.T) {
 				Drained: 3,
 				Stat:    QueueStat{Name: "orders", MessageCount: 9},
 			},
-			wants: []string{"the broker stopped delivering them"},
+			// No counter explains the remainder, so the message must name the
+			// broker-side stall and point somewhere, not just say it gave up.
+			wants: []string{"all 9 are deliverable now", "broker-side stall", "check the broker log"},
+		},
+		{
+			// The counter advertises a backlog a scan cannot find: the messages
+			// are not stuck, they are not there. Must not read as a stall.
+			name: "counter drift: scan finds nothing the counter promised",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 158782},
+				Scanned: true,
+				Counted: 0,
+			},
+			wants: []string{
+				"messageCount reports 158782 but a countMessages scan finds 0",
+				"counter has drifted",
+				"do not exist",
+				"restart the broker",
+			},
+		},
+		{
+			name: "scan confirms the backlog is real: broker is stalled",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 158782},
+				Scanned: true,
+				Counted: 158782,
+			},
+			wants: []string{"a scan confirms 158782 real message(s)", "broker-side stall"},
+		},
+		{
+			// The scan itself failed, so Counted says nothing: fall back to the
+			// plain stall wording rather than claim drift on a zero value.
+			name: "no scan ran: falls back to the stall wording",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 9},
+				Scanned: false,
+			},
+			wants: []string{"all 9 are deliverable now", "broker-side stall"},
+		},
+		{
+			// A real reason always wins over the scan-derived wording.
+			name: "scheduled remainder still reports the counter reason",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 2,
+				Stat:    QueueStat{Name: "orders", MessageCount: 5, ScheduledCount: 5},
+				Scanned: true,
+				Counted: 5,
+			},
+			wants: []string{"5 scheduled for later delivery"},
 		},
 	}
 	for _, tc := range tests {
diff --git a/internal/broker/drift.go b/internal/broker/drift.go
new file mode 100644
index 0000000..3258311
--- /dev/null
+++ b/internal/broker/drift.go
@@ -0,0 +1,152 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// DriftVerdict is what a drift check concluded about one queue.
+type DriftVerdict int
+
+const (
+	// DriftNone: the scan and the counter agree. The depth is real.
+	DriftNone DriftVerdict = iota
+	// DriftConfirmed: the counter is stable and still disagrees with a scan of
+	// the queue. The counter is advertising messages that are not there.
+	DriftConfirmed
+	// DriftInconclusive: the queue's depth moved while it was being checked, so
+	// the counter and the scan were sampled against different queues and their
+	// disagreement proves nothing.
+	DriftInconclusive
+)
+
+func (v DriftVerdict) String() string {
+	switch v {
+	case DriftConfirmed:
+		return "DRIFT"
+	case DriftInconclusive:
+		return "inconclusive"
+	default:
+		return "ok"
+	}
+}
+
+// DriftReport is one queue's drift check: the counter read on either side of a
+// scan, and what the scan found in between.
+//
+// The counter is read twice on purpose. A drift check compares two numbers
+// sampled at different instants, so on a queue with live producers or consumers
+// they disagree simply because the queue moved -- which would report drift on
+// every healthy busy queue. Bracketing the scan with two counter reads
+// distinguishes the two: only a counter that did not move while the scan ran
+// can be meaningfully compared against it.
+type DriftReport struct {
+	Queue string
+
+	// CounterBefore and CounterAfter are QueueStat.MessageCount either side of
+	// the scan; Counted is what the scan found.
+	CounterBefore int64
+	Counted       int64
+	CounterAfter  int64
+}
+
+// Verdict classifies the report. Missing is the magnitude of a confirmed drift:
+// how many messages the counter claims that do not exist.
+func (r DriftReport) Verdict() DriftVerdict {
+	if r.CounterBefore != r.CounterAfter {
+		return DriftInconclusive
+	}
+	if r.CounterBefore == r.Counted {
+		return DriftNone
+	}
+	return DriftConfirmed
+}
+
+// Missing is how many messages the counter over-reports. It is only meaningful
+// for a DriftConfirmed report; it is 0 otherwise, including when a scan finds
+// MORE than the counter claims (an under-reporting counter is a different bug
+// and does not strand messages, so it is not what this reports).
+func (r DriftReport) Missing() int64 {
+	if r.Verdict() != DriftConfirmed {
+		return 0
+	}
+	if n := r.CounterBefore - r.Counted; n > 0 {
+		return n
+	}
+	return 0
+}
+
+// CheckDrift reports whether a queue's message counter agrees with a scan of
+// the queue itself.
+//
+// This detects the failure that makes a queue undrainable: for a paged queue
+// the broker derives messageCount from page-counter journal records, and when
+// those drift the queue advertises a depth whose messages do not exist. It
+// reports a large backlog, refuses to deliver any of it, and no export can ever
+// empty it -- the only repair is a broker restart, which rebuilds the counter
+// from the journal.
+//
+// It is deliberately not free: the scan walks the queue, so this costs far more
+// than reading the counter and is meant to be run when a queue looks wrong, not
+// on every status.
+func (c *Client) CheckDrift(ctx context.Context, name string) (DriftReport, error) {
+	before, err := c.QueueStatByName(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	counted, err := c.CountMessages(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	after, err := c.QueueStatByName(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	return DriftReport{
+		Queue:         name,
+		CounterBefore: before.MessageCount,
+		Counted:       counted,
+		CounterAfter:  after.MessageCount,
+	}, nil
+}
+
+// isQueueGone reports whether err means "that queue no longer exists".
+//
+// The two calls a drift check makes report it differently: listQueues simply
+// does not return the queue, which QueueStatByName turns into ErrQueueNotFound,
+// while a management op against a queue that is gone is REJECTED by the broker
+// with "Cannot find resource with name queue.X" -- an error with no structure to
+// match on, hence the string. Matching it is worth it: an auto-delete race is
+// routine and must not fail a whole drift sweep.
+func isQueueGone(err error) bool {
+	if errors.Is(err, ErrQueueNotFound) {
+		return true
+	}
+	return strings.Contains(err.Error(), "Cannot find resource")
+}
+
+// CheckDriftAll runs CheckDrift over every user queue.
+//
+// A queue that vanishes mid-check (auto-deleted once emptied) is skipped rather
+// than failing the run: it holds nothing, so it cannot be hiding a drifted
+// backlog.
+func (c *Client) CheckDriftAll(ctx context.Context) ([]DriftReport, error) {
+	queues, err := c.ListQueues(ctx)
+	if err != nil {
+		return nil, err
+	}
+	reports := make([]DriftReport, 0, len(queues))
+	for _, q := range queues {
+		r, err := c.CheckDrift(ctx, q.Name)
+		if err != nil {
+			if isQueueGone(err) {
+				continue
+			}
+			return nil, fmt.Errorf("check drift on %s: %w", q.Name, err)
+		}
+		reports = append(reports, r)
+	}
+	return reports, nil
+}
diff --git a/internal/broker/drift_integration_test.go b/internal/broker/drift_integration_test.go
new file mode 100644
index 0000000..7801be3
--- /dev/null
+++ b/internal/broker/drift_integration_test.go
@@ -0,0 +1,126 @@
+package broker
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestCheckDriftOnHealthyQueue pins the direction that matters most: a drift
+// check must NOT cry drift on a queue whose counter is fine. A detector that
+// reports drift on healthy queues is worse than none, because it trains the
+// operator to ignore the one queue that has actually drifted.
+func TestCheckDriftOnHealthyQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-healthy")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(200, 128, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	r, err := c.CheckDrift(ctx, queue)
+	if err != nil {
+		t.Fatalf("CheckDrift: %v", err)
+	}
+	if r.Verdict() != DriftNone {
+		t.Fatalf("verdict = %v on a healthy queue, want DriftNone; report=%+v", r.Verdict(), r)
+	}
+	if r.Counted != 200 || r.CounterBefore != 200 {
+		t.Fatalf("report = %+v, want counter and scan both 200", r)
+	}
+	if r.Missing() != 0 {
+		t.Fatalf("Missing() = %d on a healthy queue, want 0", r.Missing())
+	}
+}
+
+// TestCheckDriftCountsMessagesNoConsumerCanTake guards against a false positive
+// that would otherwise be easy to ship: scheduled messages are real messages on
+// the queue that no consumer can receive yet. The scan counts them and so does
+// the counter, so this is NOT drift -- only DeliverableNow() should exclude them.
+func TestCheckDriftCountsMessagesNoConsumerCanTake(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-scheduled")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	msgs := GenerateMessages(40, 128, nil)
+	scheduleAt(msgs, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, msgs, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	r, err := c.CheckDrift(ctx, queue)
+	if err != nil {
+		t.Fatalf("CheckDrift: %v", err)
+	}
+	if r.Verdict() != DriftNone {
+		t.Fatalf("verdict = %v on a wholly scheduled queue, want DriftNone; report=%+v", r.Verdict(), r)
+	}
+	if r.Counted != 40 {
+		t.Fatalf("scan counted %d, want 40: scheduled messages are still on the queue", r.Counted)
+	}
+}
+
+// TestCheckDriftAllSkipsInternalQueues pins that a drift sweep covers the user's
+// queues and does not fail on the broker's own.
+func TestCheckDriftAllSkipsInternalQueues(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-sweep")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+	if _, err := c.Produce(ctx, queue, GenerateMessages(10, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	reports, err := c.CheckDriftAll(ctx)
+	if err != nil {
+		t.Fatalf("CheckDriftAll: %v", err)
+	}
+	var found bool
+	for _, r := range reports {
+		if r.Queue == queue {
+			found = true
+			if r.Verdict() != DriftNone {
+				t.Fatalf("verdict = %v for %s, want DriftNone; report=%+v", r.Verdict(), queue, r)
+			}
+		}
+	}
+	if !found {
+		t.Fatalf("sweep of %d queue(s) did not include %s", len(reports), queue)
+	}
+}
diff --git a/internal/broker/drift_test.go b/internal/broker/drift_test.go
new file mode 100644
index 0000000..7d73e2c
--- /dev/null
+++ b/internal/broker/drift_test.go
@@ -0,0 +1,113 @@
+package broker
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestDriftReportVerdict(t *testing.T) {
+	tests := []struct {
+		name        string
+		report      DriftReport
+		want        DriftVerdict
+		wantMissing int64
+	}{
+		{
+			name:   "counter agrees with the scan",
+			report: DriftReport{CounterBefore: 500, Counted: 500, CounterAfter: 500},
+			want:   DriftNone,
+		},
+		{
+			name:   "empty queue agrees trivially",
+			report: DriftReport{},
+			want:   DriftNone,
+		},
+		{
+			// The incident: the counter advertises a backlog no scan can find.
+			name:        "stable counter promising messages a scan cannot find",
+			report:      DriftReport{CounterBefore: 158782, Counted: 0, CounterAfter: 158782},
+			want:        DriftConfirmed,
+			wantMissing: 158782,
+		},
+		{
+			name:        "partial drift still reports the gap",
+			report:      DriftReport{CounterBefore: 1000, Counted: 400, CounterAfter: 1000},
+			want:        DriftConfirmed,
+			wantMissing: 600,
+		},
+		{
+			// A live queue moves between the samples, so the counter and the scan
+			// disagree for an ordinary reason. Reporting drift here would light up
+			// every healthy queue that has a consumer attached.
+			name:   "queue draining under us is inconclusive, not drift",
+			report: DriftReport{CounterBefore: 1000, Counted: 940, CounterAfter: 880},
+			want:   DriftInconclusive,
+		},
+		{
+			name:   "queue filling under us is inconclusive",
+			report: DriftReport{CounterBefore: 100, Counted: 130, CounterAfter: 160},
+			want:   DriftInconclusive,
+		},
+		{
+			// The counter under-reporting does not strand messages: the scan can
+			// still find them and a drain can still take them.
+			name:        "scan finding more than the counter claims reports no missing messages",
+			report:      DriftReport{CounterBefore: 10, Counted: 25, CounterAfter: 10},
+			want:        DriftConfirmed,
+			wantMissing: 0,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.report.Verdict(); got != tc.want {
+				t.Fatalf("Verdict() = %v, want %v", got, tc.want)
+			}
+			if got := tc.report.Missing(); got != tc.wantMissing {
+				t.Fatalf("Missing() = %d, want %d", got, tc.wantMissing)
+			}
+		})
+	}
+}
+
+func TestDriftVerdictString(t *testing.T) {
+	tests := []struct {
+		verdict DriftVerdict
+		want    string
+	}{
+		{DriftNone, "ok"},
+		{DriftConfirmed, "DRIFT"},
+		{DriftInconclusive, "inconclusive"},
+	}
+	for _, tc := range tests {
+		if got := tc.verdict.String(); got != tc.want {
+			t.Fatalf("String() = %q, want %q", got, tc.want)
+		}
+	}
+}
+
+// A queue auto-deleted mid-sweep must not fail a whole drift check, and the two
+// calls a check makes report it in two different ways.
+func TestIsQueueGone(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"QueueStatByName not-found", fmt.Errorf("%w: orders", ErrQueueNotFound), true},
+		{"wrapped not-found", fmt.Errorf("check drift: %w", fmt.Errorf("%w: orders", ErrQueueNotFound)), true},
+		{
+			name: "broker rejection for a missing management resource",
+			err:  errors.New(`broker rejected queue.orders.countMessages: Cannot find resource with name queue.orders`),
+			want: true,
+		},
+		{"an unrelated failure is not a missing queue", errors.New("connection reset by peer"), false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isQueueGone(tc.err); got != tc.want {
+				t.Fatalf("isQueueGone(%v) = %t, want %t", tc.err, got, tc.want)
+			}
+		})
+	}
+}
diff --git a/internal/broker/management.go b/internal/broker/management.go
index 603e996..d99aaf7 100644
--- a/internal/broker/management.go
+++ b/internal/broker/management.go
@@ -27,6 +27,13 @@ type QueueStat struct {
 	DeliveringCount int64  `json:"deliveringCount,string"`
 	ConsumerCount   int64  `json:"consumerCount,string"`
 	Paused          bool   `json:"paused,string"`
+
+	// Temporary and Internal are the broker's own classification of the queue.
+	// They are how ListQueues recognises queues that are not the user's data
+	// (dynamic reply queues, internal bookkeeping) without having to guess from
+	// the name.
+	Temporary bool `json:"temporary,string"`
+	Internal  bool `json:"internalQueue,string"`
 }
 
 // DeliverableNow reports how many of the queue's messages the broker would
@@ -150,17 +157,39 @@ func (c *Client) PurgeQueue(ctx context.Context, name string) (int64, error) {
 	if err != nil {
 		return 0, fmt.Errorf("purge %s: %w", name, err)
 	}
-	// The reply value is the removed count, encoded as a JSON array holding a
-	// single number ("[12]"), matching the other management replies' shape.
+	return parseCountReply(reply), nil
+}
+
+// CountMessages returns how many messages a scan of the queue actually finds.
+//
+// This is NOT the same as QueueStat.MessageCount. MessageCount is a counter the
+// broker maintains (for a paged queue, derived from page-counter journal
+// records); countMessages walks the queue itself. When the two disagree, the
+// counter has drifted and reports messages that do not exist -- a queue that
+// advertises a large depth, refuses to deliver anything, and can never be
+// drained. Comparing the two is the only way to tell that apart from a broker
+// that genuinely holds messages but has stalled.
+func (c *Client) CountMessages(ctx context.Context, name string) (int64, error) {
+	reply, err := c.callManagement(ctx, "queue."+name, "countMessages", "[]")
+	if err != nil {
+		return 0, fmt.Errorf("count messages on %s: %w", name, err)
+	}
+	return parseCountReply(reply), nil
+}
+
+// parseCountReply decodes the count-shaped management reply the broker returns
+// for countMessages and removeAllMessages: a JSON array holding a single number,
+// encoded as a string ("[12]").
+func parseCountReply(reply *amqp.Message) int64 {
 	raw, ok := reply.Value.(string)
 	if !ok {
-		return 0, nil
+		return 0
 	}
 	var outer []int64
 	if err := json.Unmarshal([]byte(raw), &outer); err != nil || len(outer) == 0 {
-		return 0, nil
+		return 0
 	}
-	return outer[0], nil
+	return outer[0]
 }
 
 func parseQueueStatsReply(reply *amqp.Message) ([]QueueStat, error) {
@@ -195,10 +224,22 @@ func parseQueueStatsRaw(reply *amqp.Message) ([]QueueStat, error) {
 	return paged.Data, nil
 }
 
+// filterInternalQueues drops the queues an export has no business draining:
+// the broker's own internal and temporary queues (including the dynamic reply
+// queue callManagement opens for every management call).
+//
+// It trusts the broker's temporary/internalQueue flags rather than the shape of
+// the name. An earlier version also skipped every 36-character name to catch
+// UUID-named reply queues, which silently excluded any real user queue whose
+// name happened to be 36 characters long -- an export that quietly skips a
+// queue is exactly the kind of silent data loss this tool exists to prevent.
 func filterInternalQueues(in []QueueStat) []QueueStat {
 	var out []QueueStat
 	for _, q := range in {
-		if strings.HasPrefix(q.Name, "activemq.") || strings.HasPrefix(q.Name, "$") || len(q.Name) == 36 {
+		if q.Temporary || q.Internal {
+			continue
+		}
+		if strings.HasPrefix(q.Name, "activemq.") || strings.HasPrefix(q.Name, "$") {
 			continue
 		}
 		out = append(out, q)
diff --git a/internal/broker/management_test.go b/internal/broker/management_test.go
index a4be2ca..3104bdb 100644
--- a/internal/broker/management_test.go
+++ b/internal/broker/management_test.go
@@ -79,11 +79,23 @@ func TestFilterInternalQueues(t *testing.T) {
 		{Name: "orders", MessageCount: 5},
 		{Name: "activemq.notifications", MessageCount: 1},
 		{Name: "$sys.foo", MessageCount: 1},
-		{Name: "123e4567-e89b-12d3-a456-426614174000", MessageCount: 1}, // 36 chars
+		// The dynamic reply queue callManagement opens: the broker flags it
+		// temporary, which is what we filter on.
+		{Name: "123e4567-e89b-12d3-a456-426614174000", MessageCount: 1, Temporary: true},
+		{Name: "internal.bookkeeping", MessageCount: 1, Internal: true},
+		// A real user queue that is 36 characters long must survive: an export
+		// that silently skips a queue loses every message on it.
+		{Name: "billing_credit_approved_debit_rows_q", MessageCount: 7},
 		{Name: "payments", MessageCount: 2},
 	}
 	got := filterInternalQueues(in)
-	if len(got) != 2 || got[0].Name != "orders" || got[1].Name != "payments" {
-		t.Fatalf("unexpected filter result: %+v", got)
+	want := []string{"orders", "billing_credit_approved_debit_rows_q", "payments"}
+	if len(got) != len(want) {
+		t.Fatalf("filterInternalQueues returned %d queues, want %d: %+v", len(got), len(want), got)
+	}
+	for i, w := range want {
+		if got[i].Name != w {
+			t.Fatalf("filterInternalQueues()[%d] = %q, want %q", i, got[i].Name, w)
+		}
 	}
 }
diff --git a/internal/cli/commands_integration_test.go b/internal/cli/commands_integration_test.go
index 1151e80..450bfaf 100644
--- a/internal/cli/commands_integration_test.go
+++ b/internal/cli/commands_integration_test.go
@@ -40,6 +40,35 @@ func TestCommandsAgainstBroker(t *testing.T) {
 		if !strings.Contains(out, "orders") {
 			t.Fatalf("status output missing queue: %q", out)
 		}
+		// A depth alone cannot explain why an export leaves messages behind;
+		// these columns are what let an operator tell "the broker is holding
+		// them back" from "the broker is stalled".
+		for _, col := range []string{"MESSAGE COUNT", "DELIVERABLE", "SCHEDULED", "DELIVERING", "CONSUMERS", "PAUSED"} {
+			if !strings.Contains(out, col) {
+				t.Fatalf("status output missing the %q column: %q", col, out)
+			}
+		}
+	})
+
+	// The broker under test is healthy, so --verify must come back clean and
+	// exit 0. The flag exists to single out a queue whose counter lies; one that
+	// flags a healthy broker would just be noise an operator learns to ignore.
+	t.Run("status --verify on a healthy broker", func(t *testing.T) {
+		out, err := run(t, "status", "--verify")
+		if err != nil {
+			t.Fatalf("status --verify on a healthy broker: %v\n%s", err, out)
+		}
+		for _, col := range []string{"COUNTER", "SCANNED", "MISSING", "VERDICT"} {
+			if !strings.Contains(out, col) {
+				t.Fatalf("status --verify output missing the %q column: %q", col, out)
+			}
+		}
+		if !strings.Contains(out, "orders") {
+			t.Fatalf("status --verify output missing queue: %q", out)
+		}
+		if strings.Contains(out, "DRIFT") {
+			t.Fatalf("status --verify reported drift on a healthy broker: %q", out)
+		}
 	})
 
 	t.Run("health", func(t *testing.T) {
diff --git a/internal/cli/status.go b/internal/cli/status.go
index 16ea890..4fb0f1e 100644
--- a/internal/cli/status.go
+++ b/internal/cli/status.go
@@ -10,7 +10,7 @@ import (
 )
 
 func newStatusCmd() *cobra.Command {
-	return &cobra.Command{
+	cmd := &cobra.Command{
 		Use:   "status",
 		Short: "List queues and message counts (descending)",
 		RunE: func(cmd *cobra.Command, _ []string) error {
@@ -21,16 +21,69 @@ func newStatusCmd() *cobra.Command {
 				return err
 			}
 			defer c.Close(ctx)
+			if verify, _ := cmd.Flags().GetBool("verify"); verify {
+				return runStatusVerify(cmd, c)
+			}
 			stats, err := c.ListQueues(ctx)
 			if err != nil {
 				return err
 			}
 			w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 3, ' ', 0)
-			fmt.Fprintln(w, "QUEUE NAME\tMESSAGE COUNT")
+			// MESSAGE COUNT alone cannot explain why an export leaves messages
+			// behind: the depth also counts messages the broker will not hand to
+			// a consumer. DELIVERABLE is the subset a drain can actually take,
+			// and the remaining columns say where the difference went.
+			fmt.Fprintln(w, "QUEUE NAME\tMESSAGE COUNT\tDELIVERABLE\tSCHEDULED\tDELIVERING\tCONSUMERS\tPAUSED")
 			for _, s := range stats {
-				fmt.Fprintf(w, "%s\t%d\n", s.Name, s.MessageCount)
+				fmt.Fprintf(w, "%s\t%d\t%d\t%d\t%d\t%d\t%t\n",
+					s.Name, s.MessageCount, s.DeliverableNow(),
+					s.ScheduledCount, s.DeliveringCount, s.ConsumerCount, s.Paused)
 			}
 			return w.Flush()
 		},
 	}
+	cmd.Flags().Bool("verify", false,
+		"scan every queue and report drift detection (slow: walks each queue)")
+	return cmd
+}
+
+// runStatusVerify prints the drift check: the counter against a scan of each
+// queue. It exits non-zero on confirmed drift, because a drifted counter means
+// the broker is advertising messages that do not exist and an export of that
+// queue can never complete -- a condition a script should be able to catch.
+func runStatusVerify(cmd *cobra.Command, c *broker.Client) error {
+	// The scan walks every queue, which on a deep backlog takes far longer than
+	// the --timeout meant for connecting, so this runs under a signal context
+	// like the other long operations rather than being cut off mid-sweep.
+	ctx, stop := signalCtx()
+	defer stop()
+
+	reports, err := c.CheckDriftAll(ctx)
+	if err != nil {
+		return err
+	}
+	w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 3, ' ', 0)
+	// COUNTER is what the broker advertises; SCANNED is what walking the queue
+	// actually finds. MISSING is the gap: messages the counter promises that are
+	// not there.
+	fmt.Fprintln(w, "QUEUE NAME\tCOUNTER\tSCANNED\tMISSING\tVERDICT")
+	drifted := 0
+	for _, r := range reports {
+		if r.Verdict() == broker.DriftConfirmed {
+			drifted++
+		}
+		fmt.Fprintf(w, "%s\t%d\t%d\t%d\t%s\n",
+			r.Queue, r.CounterBefore, r.Counted, r.Missing(), r.Verdict())
+	}
+	if err := w.Flush(); err != nil {
+		return err
+	}
+	if drifted > 0 {
+		fmt.Fprintf(cmd.OutOrStdout(),
+			"\n%d queue(s) have a drifted message counter: they advertise messages that do not exist,\n"+
+				"so no export can ever empty them. Restart the broker to rebuild the counters from its journal.\n",
+			drifted)
+		return fmt.Errorf("%d queue(s) with a drifted message counter", drifted)
+	}
+	return nil
 }

From 62b8c46284c8924b3f4bfd438a7ed0a540ec0eef Mon Sep 17 00:00:00 2001
From: Richard Martikan <ric.martikan@gmail.com>
Date: Thu, 16 Jul 2026 21:11:34 +0200
Subject: [PATCH 3/4] add fmt/vet for Makefile tests

---
 Makefile                 | 20 +++++++++++++++-----
 internal/broker/drain.go |  2 --
 2 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/Makefile b/Makefile
index 3778351..924afa0 100644
--- a/Makefile
+++ b/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"
diff --git a/internal/broker/drain.go b/internal/broker/drain.go
index 0f71954..cb0c8b4 100644
--- a/internal/broker/drain.go
+++ b/internal/broker/drain.go
@@ -42,7 +42,6 @@ type PartialDrainError struct {
 	Counted int64
 }
 
-
 // outcome is what the broker's counters say about a finished drain pass.
 type outcome int
 
@@ -166,7 +165,6 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 	}
 }
 
-
 // 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

From 410990de4997a8508abf88e4b5c8256cea5e064b Mon Sep 17 00:00:00 2001
From: Richard Martikan <ric.martikan@gmail.com>
Date: Fri, 17 Jul 2026 08:16:15 +0200
Subject: [PATCH 4/4] Fix integration test of helth-report

---
 internal/broker/health_integration_test.go | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/internal/broker/health_integration_test.go b/internal/broker/health_integration_test.go
index 6939348..d69fe25 100644
--- a/internal/broker/health_integration_test.go
+++ b/internal/broker/health_integration_test.go
@@ -22,8 +22,14 @@ func TestCheckHealthIntegration(t *testing.T) {
 	if err != nil {
 		t.Fatalf("health: %v", err)
 	}
-	if h.Verdict != OK {
-		t.Fatalf("fresh broker should be OK, got %s (disk %.1f mem %.1f)", h.Verdict, h.DiskUsagePct, h.MemoryUsagePct)
+	// DiskUsagePct reflects the real filesystem holding the broker's data dir,
+	// which on a dev host is a shared partition this test does not control. A
+	// busy disk (>=70%) legitimately yields DEGRADED on an otherwise-fresh
+	// broker, so we do NOT assert Verdict == OK. What must always hold for a
+	// fresh broker is that it is not in the CRITICAL band (disk/mem >90% or
+	// producers blocked).
+	if h.Verdict == Critical {
+		t.Fatalf("fresh broker should not be CRITICAL, got %s (disk %.1f mem %.1f blocking %v)", h.Verdict, h.DiskUsagePct, h.MemoryUsagePct, h.Blocking)
 	}
 	// I1: a fresh broker with a near-empty disk is well below max-disk-usage,
 	// so the disk-full producer-block indicator must be false.