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

Author: bootjp

internal/backup/sqs.go

@@ -252,18 +252,23 @@ func (s *SQSEncoder) HandleSideRecord(prefix string, key, value []byte) error {
 }
 
 // Finalize flushes every queue's _queue.json and messages.jsonl. Queues
-// with buffered messages but no meta record (orphans) emit a warning and
-// are skipped — restoring orphan messages without a queue config would
-// silently create a queue with default settings, which is rarely what
-// the operator wants.
+// with buffered messages but no meta record (orphans) emit a warning
+// and have their messages dropped — restoring orphan messages without
+// a queue config would silently create a queue with default settings,
+// which is rarely what the operator wants. However, if
+// --include-sqs-side-records is on and this orphan queue has buffered
+// side records (vis/byage/dedup/group/tombstone), those are still
+// flushed under the encoded-prefix directory: the most common reason
+// for a missing meta is a DeleteQueue that left tombstones, and
+// dropping exactly those records is the opposite of what the operator
+// asked for. Codex P2 round 8.
 func (s *SQSEncoder) Finalize() error {
 	var firstErr error
 	for _, st := range s.queues {
 		if st.meta == nil {
-			s.emitWarn("sqs_orphan_messages",
-				"encoded_queue", st.encoded,
-				"buffered_messages", len(st.messages),
-				"hint", "no !sqs|queue|meta record matched this encoded prefix; messages dropped from the dump")
+			if err := s.flushOrphanQueueSideRecords(st); err != nil && firstErr == nil {
+				firstErr = err
+			}
 			continue
 		}
 		if err := s.flushQueue(st); err != nil && firstErr == nil {
@@ -273,6 +278,30 @@ func (s *SQSEncoder) Finalize() error {
 	return firstErr
 }
 
+// flushOrphanQueueSideRecords emits buffered side records for a queue
+// whose !sqs|queue|meta row never arrived. Without this branch,
+// --include-sqs-side-records silently drops the post-DeleteQueue
+// tombstones and dedup-window history operators most often opt in
+// for. The orphan dir is named by the encoded prefix because no
+// decoded queue name is available; restore tools can join it with
+// the messages-dropped warning to reconstruct context.
+func (s *SQSEncoder) flushOrphanQueueSideRecords(st *sqsQueueState) error {
+	s.emitWarn("sqs_orphan_messages",
+		"encoded_queue", st.encoded,
+		"buffered_messages", len(st.messages),
+		"buffered_side_records", len(st.internalBuf),
+		"hint", "no !sqs|queue|meta record matched this encoded prefix; messages dropped from the dump")
+	if !s.includeSideRecords || len(st.internalBuf) == 0 {
+		return nil
+	}
+	// Use the encoded prefix as the directory name — it's the only
+	// stable identifier available when meta is missing. Suffix it
+	// with `.orphan` so a restore tool cannot mistake it for a real
+	// queue dir produced from a successful meta flush.
+	dir := filepath.Join(s.outRoot, "sqs", st.encoded+".orphan")
+	return s.flushInternals(dir, st.internalBuf)
+}
+
 func (s *SQSEncoder) flushQueue(st *sqsQueueState) error {
 	dir := filepath.Join(s.outRoot, "sqs", EncodeSegment([]byte(st.name)))
 	if err := os.MkdirAll(dir, 0o755); err != nil { //nolint:mnd // 0755 == standard dir mode

internal/backup/sqs_test.go

@@ -410,6 +410,63 @@ func TestSQS_SideRecordsFlushedEvenWhenZeroMessages(t *testing.T) {
 	}
 }
 
+// TestSQS_OrphanQueueSideRecordsPreserved is the regression for Codex
+// P2 round 8: when a queue's !sqs|queue|meta record is missing (e.g.
+// after DeleteQueue left tombstones but the meta row was removed) and
+// --include-sqs-side-records is on, side records were silently dropped
+// alongside any orphan messages. The opt-in contract is the opposite:
+// side records exist precisely so deletion-era state is recoverable.
+// Now those records flush to a `<encoded>.orphan` directory while the
+// orphan-messages warning fires.
+func TestSQS_OrphanQueueSideRecordsPreserved(t *testing.T) {
+	t.Parallel()
+	enc, root := newSQSEncoder(t)
+	enc.WithIncludeSideRecords(true)
+	var events []string
+	enc.WithWarnSink(func(event string, _ ...any) { events = append(events, event) })
+	// Side record arrives without a meta row first (deletion-era).
+	encQueue := "ZGVsZXRlZA" // base64url("deleted")
+	visKey := append([]byte(SQSMsgVisPrefix), []byte(encQueue)...)
+	visKey = append(visKey, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01)
+	if err := enc.HandleSideRecord(SQSMsgVisPrefix, visKey, []byte("vis-record")); err != nil {
+		t.Fatal(err)
+	}
+	if err := enc.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	if len(events) == 0 || events[0] != "sqs_orphan_messages" {
+		t.Fatalf("expected sqs_orphan_messages warning, got %v", events)
+	}
+	want := filepath.Join(root, "sqs", encQueue+".orphan", "_internals", "side_records.jsonl")
+	if _, err := os.Stat(want); err != nil {
+		t.Fatalf("expected orphan side-records file at %s: %v", want, err)
+	}
+}
+
+// TestSQS_OrphanQueueSideRecordsSuppressedWhenOptOut asserts that the
+// orphan-side-records branch is gated on --include-sqs-side-records:
+// without the flag, the warning still fires but no .orphan dir is
+// created (consistent with the default-off contract for side records).
+func TestSQS_OrphanQueueSideRecordsSuppressedWhenOptOut(t *testing.T) {
+	t.Parallel()
+	enc, root := newSQSEncoder(t)
+	// includeSideRecords is off by default — HandleSideRecord drops
+	// the record at intake, so the buffer is empty by Finalize. We
+	// exercise the path anyway to confirm no .orphan dir is created.
+	encQueue := "b3B0LW91dA" // base64url("opt-out")
+	visKey := append([]byte(SQSMsgVisPrefix), []byte(encQueue)...)
+	visKey = append(visKey, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02)
+	if err := enc.HandleSideRecord(SQSMsgVisPrefix, visKey, []byte("vis")); err != nil {
+		t.Fatal(err)
+	}
+	if err := enc.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(filepath.Join(root, "sqs", encQueue+".orphan")); !os.IsNotExist(err) {
+		t.Fatalf("orphan dir created without --include-sqs-side-records: stat err=%v", err)
+	}
+}
+
 func TestSQS_DefaultDoesNotEmitInternals(t *testing.T) {
 	t.Parallel()
 	enc, root := newSQSEncoder(t)