Merge remote-tracking branch 'origin/feat/backup-phase0a-sqs' into feat/backup-phase0a-s3

Author: bootjp

internal/backup/disk_full_other.go

@@ -0,0 +1,15 @@
+//go:build !unix && !windows
+
+package backup
+
+// isDiskFullError is the fallback for non-unix/non-windows targets
+// (js, wasip1, plan9). Those runtimes either do not surface a
+// disk-full errno through Go's syscall package or do not have a
+// meaningful disk concept (wasm with no host filesystem, plan9 with
+// its own error vocabulary). Returning false matches the documented
+// helper contract: callers treat unrecognised errors as
+// non-retriable, which is the safe default. Codex P2 round 10.
+func isDiskFullError(err error) bool {
+	_ = err
+	return false
+}

internal/backup/open_nofollow_other.go

@@ -0,0 +1,25 @@
+//go:build !unix && !windows
+
+package backup
+
+import (
+	"os"
+
+	cockroachdberr "github.com/cockroachdb/errors"
+)
+
+// openSidecarFile is the fallback for non-unix/non-windows targets
+// (js, wasip1, plan9). syscall.O_NOFOLLOW and the unix nlink-check
+// path are unavailable; we keep a Lstat-then-OpenFile guard to at
+// least refuse pre-existing symlinks. The remaining TOCTOU window
+// is acceptable here because dump tooling on those targets is
+// offline / sandboxed and the threat model that motivated the unix
+// hardening (a local adversary swapping the path between syscalls)
+// does not apply. Codex P2 round 10.
+func openSidecarFile(path string) (*os.File, error) {
+	if info, err := os.Lstat(path); err == nil && info.Mode()&os.ModeSymlink != 0 {
+		return nil, cockroachdberr.WithStack(cockroachdberr.Newf(
+			"backup: refusing to overwrite symlink at %s", path))
+	}
+	return os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) //nolint:gosec,mnd // path is composed from output-root + fixed file name; 0600 is the standard owner-only mode
+}

internal/backup/sqs.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
 	"unicode/utf8"
 
@@ -108,6 +109,18 @@ type sqsQueueState struct {
 	name     string // decoded queue name; populated on meta arrival
 	meta     *sqsQueueMetaPublic
 	messages []sqsMessageRecord
+	// activeGen captures the queue's current generation, parsed
+	// from a !sqs|queue|gen|<encoded> record. PurgeQueue and
+	// DeleteQueue bump the generation so the reaper can later
+	// drop residual !sqs|msg|data|<queue><oldGen><...> rows; if
+	// the snapshot is taken before reaping completes those stale
+	// rows are still in the keyspace and must NOT be emitted to
+	// messages.jsonl. Zero means "no gen record observed yet" and
+	// disables the filter (the live cluster always writes a gen
+	// record on CreateQueue, so a missing gen at backup time
+	// means we have an orphan-only queue and the orphan branch
+	// already drops messages). Codex P1 round 10.
+	activeGen uint64
 	// internalBuf accumulates side records in their on-disk shape if
 	// includeSideRecords is on. Each line is the encoded prefix +
 	// hex(rest-of-key) + value (b64) — implementation-grade detail
@@ -254,6 +267,28 @@ func (s *SQSEncoder) HandleQueueMeta(key, value []byte) error {
 	return nil
 }
 
+// HandleQueueGen processes one !sqs|queue|gen|<encoded> record. The
+// value is a base-10 decimal string holding the queue's current
+// generation (mirrors adapter/sqs_catalog.go's CreateQueue Put: the
+// live cluster writes strconv.FormatUint(gen, 10)). Capturing
+// activeGen lets flushQueue drop messages tagged with older
+// generations — those are residual rows left by PurgeQueue /
+// DeleteQueue that the reaper has not yet cleaned, and emitting
+// them to messages.jsonl would resurrect purged messages on
+// restore. Codex P1 round 10.
+func (s *SQSEncoder) HandleQueueGen(key, value []byte) error {
+	encoded, err := stripPrefixSegment(key, []byte(SQSQueueGenPrefix))
+	if err != nil {
+		return err
+	}
+	gen, err := strconv.ParseUint(strings.TrimSpace(string(value)), 10, 64) //nolint:mnd // 10 == decimal radix
+	if err != nil {
+		return errors.Wrap(ErrSQSMalformedKey, err.Error())
+	}
+	s.queueState(encoded).activeGen = gen
+	return nil
+}
+
 // HandleMessageData processes one !sqs|msg|data|<encQueue><gen><encMsgID>
 // record. The encoded queue segment is parsed out of the key and used as
 // the per-queue routing key; the message is buffered until Finalize so it
@@ -359,19 +394,24 @@ func (s *SQSEncoder) flushQueue(st *sqsQueueState) error {
 	if err := writeFileAtomic(filepath.Join(dir, "_queue.json"), mustMarshalIndent(st.meta)); err != nil {
 		return err
 	}
-	if len(st.messages) > 0 {
-		sortMessagesForEmit(st.messages)
-		jl, err := openJSONL(filepath.Join(dir, "messages.jsonl"))
-		if err != nil {
-			return err
-		}
-		for i := range st.messages {
-			if err := jl.enc.Encode(st.messages[i]); err != nil {
-				_ = closeJSONL(jl)
-				return errors.WithStack(err)
-			}
-		}
-		if err := closeJSONL(jl); err != nil {
+	// Drop messages tagged with stale generations: PurgeQueue and
+	// DeleteQueue bump the queue's gen but the reaper deletes the
+	// affected !sqs|msg|data| rows asynchronously. A snapshot taken
+	// mid-reap would otherwise resurrect purged messages on restore.
+	// activeGen == 0 means we did not see a !sqs|queue|gen| record
+	// for this queue; preserving the legacy behaviour (no filter)
+	// is the safe fallback because every CreateQueue writes a gen.
+	// Codex P1 round 10.
+	visibleMessages, dropped := filterStaleGenMessages(st.messages, st.activeGen)
+	if dropped > 0 {
+		s.emitWarn("sqs_stale_generation_messages_dropped",
+			"queue", st.name,
+			"active_gen", st.activeGen,
+			"dropped", dropped,
+			"hint", "messages with mismatched queue_generation suppressed; reaper had not finished cleanup at snapshot time")
+	}
+	if len(visibleMessages) > 0 {
+		if err := writeMessagesJSONL(dir, visibleMessages); err != nil {
 			return err
 		}
 	}
@@ -677,6 +717,50 @@ func decodeSQSMessageValue(value []byte) (sqsMessageRecord, error) {
 	}, nil
 }
 
+// writeMessagesJSONL emits the buffered (visible) messages to
+// messages.jsonl in send-order. Split out of flushQueue to keep that
+// function's cyclomatic complexity under the project's linter ceiling.
+func writeMessagesJSONL(dir string, msgs []sqsMessageRecord) error {
+	sortMessagesForEmit(msgs)
+	jl, err := openJSONL(filepath.Join(dir, "messages.jsonl"))
+	if err != nil {
+		return err
+	}
+	for i := range msgs {
+		if err := jl.enc.Encode(msgs[i]); err != nil {
+			_ = closeJSONL(jl)
+			return errors.WithStack(err)
+		}
+	}
+	if err := closeJSONL(jl); err != nil {
+		return err
+	}
+	return nil
+}
+
+// filterStaleGenMessages partitions in into (visible, droppedCount).
+// A message is visible if activeGen is zero (no gen record observed
+// for this queue, which means we cannot make a confident call) OR
+// its QueueGeneration matches activeGen. Anything else is residual
+// state from a PurgeQueue / DeleteQueue that the reaper has not yet
+// removed; emitting it would resurrect purged messages on restore.
+// Codex P1 round 10.
+func filterStaleGenMessages(in []sqsMessageRecord, activeGen uint64) ([]sqsMessageRecord, int) {
+	if activeGen == 0 {
+		return in, 0
+	}
+	out := make([]sqsMessageRecord, 0, len(in))
+	dropped := 0
+	for _, m := range in {
+		if m.QueueGeneration == activeGen {
+			out = append(out, m)
+			continue
+		}
+		dropped++
+	}
+	return out, dropped
+}
+
 func sortMessagesForEmit(msgs []sqsMessageRecord) {
 	sort.SliceStable(msgs, func(i, j int) bool {
 		a, b := msgs[i], msgs[j]

internal/backup/sqs_test.go

@@ -3,9 +3,11 @@ package backup
 import (
 	"bufio"
 	"bytes"
+	"encoding/base64"
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"slices"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -375,6 +377,94 @@ func TestSQS_ParseMessageDataKey_RejectsEmptyMsgIDSegment(t *testing.T) {
 	}
 }
 
+// TestSQS_StaleGenerationMessagesDropped is the regression for Codex
+// P1 round 10: PurgeQueue and DeleteQueue bump the queue's generation
+// but the affected !sqs|msg|data|<oldGen>|... rows are removed
+// asynchronously by the reaper. A snapshot taken mid-cleanup would
+// otherwise resurrect those purged messages on restore. The encoder
+// now consults the !sqs|queue|gen| record and drops messages whose
+// QueueGeneration does not match the active gen, emitting an
+// `sqs_stale_generation_messages_dropped` warning for visibility.
+func TestSQS_StaleGenerationMessagesDropped(t *testing.T) {
+	t.Parallel()
+	enc, root := newSQSEncoder(t)
+	var events []string
+	enc.WithWarnSink(func(event string, _ ...any) { events = append(events, event) })
+	queue := "q"
+	if err := enc.HandleQueueMeta(EncodeQueueMetaKey(queue), encodeQueueMetaValue(t, map[string]any{
+		"name": queue, "visibility_timeout_seconds": 30, "message_retention_seconds": 60,
+	})); err != nil {
+		t.Fatal(err)
+	}
+	encQueue := base64.RawURLEncoding.EncodeToString([]byte(queue))
+	genKey := append([]byte(SQSQueueGenPrefix), []byte(encQueue)...)
+	if err := enc.HandleQueueGen(genKey, []byte("7")); err != nil {
+		t.Fatal(err)
+	}
+	live := encodeMessageValue(t, map[string]any{
+		"message_id":            "live",
+		"body":                  []byte("ok"),
+		"send_timestamp_millis": 100,
+		"queue_generation":      7,
+	})
+	if err := enc.HandleMessageData(EncodeMsgDataKey(queue, 7, "live"), live); err != nil {
+		t.Fatal(err)
+	}
+	stale := encodeMessageValue(t, map[string]any{
+		"message_id":            "ghost",
+		"body":                  []byte("from-prev-gen"),
+		"send_timestamp_millis": 50,
+		"queue_generation":      6,
+	})
+	if err := enc.HandleMessageData(EncodeMsgDataKey(queue, 6, "ghost"), stale); err != nil {
+		t.Fatal(err)
+	}
+	if err := enc.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	recs := readMessagesJSONL(t, filepath.Join(root, "sqs", queue, "messages.jsonl"))
+	if len(recs) != 1 {
+		t.Fatalf("messages = %d want 1 (stale gen must drop)", len(recs))
+	}
+	if recs[0]["message_id"] != "live" {
+		t.Fatalf("survived msg = %v want live", recs[0]["message_id"])
+	}
+	if !slices.Contains(events, "sqs_stale_generation_messages_dropped") {
+		t.Fatalf("expected sqs_stale_generation_messages_dropped warning, got %v", events)
+	}
+}
+
+// TestSQS_StaleGenerationFilterDisabledWithoutGenRecord asserts the
+// safe fallback: a queue with no !sqs|queue|gen| record observed
+// keeps the legacy behavior (no filter), so a backup that lacks the
+// gen record does not silently lose every message.
+func TestSQS_StaleGenerationFilterDisabledWithoutGenRecord(t *testing.T) {
+	t.Parallel()
+	enc, root := newSQSEncoder(t)
+	queue := "q"
+	if err := enc.HandleQueueMeta(EncodeQueueMetaKey(queue), encodeQueueMetaValue(t, map[string]any{
+		"name": queue, "visibility_timeout_seconds": 30, "message_retention_seconds": 60,
+	})); err != nil {
+		t.Fatal(err)
+	}
+	val := encodeMessageValue(t, map[string]any{
+		"message_id":            "m1",
+		"body":                  []byte("payload"),
+		"send_timestamp_millis": 1000,
+		"queue_generation":      99,
+	})
+	if err := enc.HandleMessageData(EncodeMsgDataKey(queue, 99, "m1"), val); err != nil {
+		t.Fatal(err)
+	}
+	if err := enc.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	recs := readMessagesJSONL(t, filepath.Join(root, "sqs", queue, "messages.jsonl"))
+	if len(recs) != 1 {
+		t.Fatalf("messages = %d want 1 (no gen record => no filter)", len(recs))
+	}
+}
+
 // TestSQS_MessageBodyEmittedAsTextForUTF8 is the regression for Codex
 // P1 round 9 on PR #714: `body` was a `[]byte` field, so json.Encoder
 // rendered it as base64 in messages.jsonl. Replaying that JSONL via