backup: canonicalize DynamoDB N keys + sort schema arrays (PR #716, round 4)

Codex round 9 raised two issues on commit d67baf8:

1. P1: ddbKeyAttrToSegment was emitting `EncodeSegment([]byte(v.N))`
   for numeric primary-key attributes, but DynamoDB N equality is
   numeric — `"1"` and `"1.0"` (or `"100"` and `"1e2"`, or `"0.5"`
   and `"5e-1"`) name the same logical item. In migration mode
   where source and active rows used different decimal text for
   the same value, both rows survived at distinct paths and
   restore replayed duplicates. Mirror the live adapter's
   canonicalNumberString (adapter/dynamodb.go:7651) which uses
   big.Rat — same canonical form keeps backup filenames in lockstep
   with the live equality check.

2. P2: schemaToPublic ranged over Go maps for both
   global_secondary_indexes and attribute_definitions, so identical
   snapshots produced different `_schema.json` byte output across
   runs. Sort by name before append.

Tests:
  - TestDDB_CanonicalNumberKeySegment: equivalence pairs
    ("1"/"1.0", "100"/"1e2", "-0"/"0", "0.5"/"5e-1") collapse
    to the same key segment.
  - TestDDB_SchemaJSONIsDeterministic: 32 calls to schemaToPublic
    on the same schema produce identical attribute_definitions and
    GSI orders, both matching the documented sort-by-name.

Author: bootjp

internal/backup/dynamodb.go

@@ -5,8 +5,10 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"math/big"
 	"os"
 	"path/filepath"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -378,14 +380,40 @@ func ddbKeyAttrToSegment(av *pb.DynamoAttributeValue) (string, error) {
 	case *pb.DynamoAttributeValue_S:
 		return EncodeSegment([]byte(v.S)), nil
 	case *pb.DynamoAttributeValue_N:
-		return EncodeSegment([]byte(v.N)), nil
+		// DynamoDB N equality is numeric, not lexical: "1" and
+		// "1.0" name the same primary-key item, and so do "5e-1"
+		// and "0.5". Without canonicalisation each text form
+		// produces a distinct filename, so in migration mode the
+		// "active generation wins" invariant breaks (both source
+		// and active rows survive at different paths and restore
+		// replays duplicates). Mirror the live adapter's
+		// canonicalNumberString (adapter/dynamodb.go:7651) which
+		// uses big.Rat — same canonical form keeps filename
+		// identity in lockstep with the live equality check.
+		// Codex P1 round 9.
+		return EncodeSegment([]byte(canonicalDDBNumber(v.N))), nil
 	case *pb.DynamoAttributeValue_B:
 		return EncodeBinarySegment(v.B), nil
 	}
 	return "", errors.Wrapf(ErrDDBInvalidItem,
 		"primary key has unsupported attribute kind %T", av.GetValue())
 }
 
+// canonicalDDBNumber returns the canonical decimal representation of
+// a DynamoDB N value. Equivalent inputs (`"1"`, `"1.0"`, `"0.5"`,
+// `"5e-1"`, …) collapse to the same string; malformed inputs fall
+// through to a trimmed copy so a downstream parse failure surfaces
+// the original bytes rather than a silently rewritten value. The
+// implementation matches adapter/dynamodb.go canonicalNumberString
+// byte-for-byte so backup filenames track live equality.
+func canonicalDDBNumber(v string) string {
+	rat := &big.Rat{}
+	if _, ok := rat.SetString(strings.TrimSpace(v)); !ok {
+		return strings.TrimSpace(v)
+	}
+	return rat.RatString()
+}
+
 // schemaToPublic projects DynamoTableSchema into the AWS-DescribeTable
 // JSON shape documented in the design. Fields the live record carries
 // for cluster-internal reasons (key_encoding_version, generation,
@@ -407,8 +435,18 @@ func schemaToPublic(s *pb.DynamoTableSchema) ddbPublicSchema {
 	if pk.RangeKey.Name != "" {
 		pk.RangeKey.Type = defs[pk.RangeKey.Name]
 	}
-	gsis := make([]publicGSI, 0, len(s.GetGlobalSecondaryIndexes()))
-	for name, gsi := range s.GetGlobalSecondaryIndexes() {
+	// Build GSI list in deterministic name-sorted order. Ranging over
+	// the underlying map directly produced a different array order on
+	// every dump, undermining byte-for-byte reproducibility of
+	// _schema.json across runs of the same snapshot. Codex P2 round 9.
+	gsiNames := make([]string, 0, len(s.GetGlobalSecondaryIndexes()))
+	for name := range s.GetGlobalSecondaryIndexes() {
+		gsiNames = append(gsiNames, name)
+	}
+	sort.Strings(gsiNames)
+	gsis := make([]publicGSI, 0, len(gsiNames))
+	for _, name := range gsiNames {
+		gsi := s.GetGlobalSecondaryIndexes()[name]
 		g := publicGSI{
 			Name: name,
 			KeySchema: publicKeySchema{
@@ -426,9 +464,16 @@ func schemaToPublic(s *pb.DynamoTableSchema) ddbPublicSchema {
 		g.Projection.NonKeyAttributes = append([]string{}, gsi.GetProjection().GetNonKeyAttributes()...)
 		gsis = append(gsis, g)
 	}
-	attrDefs := make([]publicAttributeDefinition, 0, len(defs))
-	for name, ty := range defs {
-		attrDefs = append(attrDefs, publicAttributeDefinition{Name: name, Type: ty})
+	// AttributeDefinitions is also sorted by attribute name for the
+	// same determinism reason.
+	defNames := make([]string, 0, len(defs))
+	for name := range defs {
+		defNames = append(defNames, name)
+	}
+	sort.Strings(defNames)
+	attrDefs := make([]publicAttributeDefinition, 0, len(defNames))
+	for _, name := range defNames {
+		attrDefs = append(attrDefs, publicAttributeDefinition{Name: name, Type: defs[name]})
 	}
 	return ddbPublicSchema{
 		FormatVersion:          1,

internal/backup/dynamodb_test.go

@@ -375,6 +375,113 @@ func TestDDB_MigrationSourceGenerationItemsAreEmitted(t *testing.T) {
 	}
 }
 
+// TestDDB_CanonicalNumberKeySegment is the regression for Codex P1
+// round 9: DynamoDB N equality is numeric, not lexical, but the key
+// segment was emitted as `EncodeSegment([]byte(v.N))`. In migration
+// mode where source and active rows used different decimal text for
+// the same logical N value (e.g. "1" and "1.0"), both rows survived
+// at distinct paths and restore replayed duplicates. The encoder
+// must canonicalise via big.Rat — same canonical form as the live
+// adapter — so equivalent N literals collapse onto the same filename.
+func TestDDB_CanonicalNumberKeySegment(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		a, b string
+	}{
+		{"1", "1.0"},
+		{"100", "1e2"},
+		{"-0", "0"},
+		{"0.5", "5e-1"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.a+"_vs_"+tc.b, func(t *testing.T) {
+			t.Parallel()
+			gotA, errA := ddbKeyAttrToSegment(nAttr(tc.a))
+			gotB, errB := ddbKeyAttrToSegment(nAttr(tc.b))
+			if errA != nil || errB != nil {
+				t.Fatalf("err: %v / %v", errA, errB)
+			}
+			if gotA != gotB {
+				t.Fatalf("equivalent N values must canonicalise to the same segment: %q vs %q -> %q vs %q",
+					tc.a, tc.b, gotA, gotB)
+			}
+		})
+	}
+}
+
+// TestDDB_SchemaJSONIsDeterministic is the regression for Codex P2
+// round 9: schemaToPublic ranged over Go maps for both
+// global_secondary_indexes and attribute_definitions, so identical
+// snapshots produced different `_schema.json` byte output across
+// runs. The keys are now sorted before append.
+func TestDDB_SchemaJSONIsDeterministic(t *testing.T) {
+	t.Parallel()
+	schema := &pb.DynamoTableSchema{
+		TableName:  "t",
+		PrimaryKey: &pb.DynamoKeySchema{HashKey: "id"},
+		AttributeDefinitions: map[string]string{
+			"zeta": "S", "alpha": "S", "id": "S", "mu": "N",
+		},
+		GlobalSecondaryIndexes: map[string]*pb.DynamoGlobalSecondaryIndex{
+			"gZ": {KeySchema: &pb.DynamoKeySchema{HashKey: "zeta"}, Projection: &pb.DynamoGSIProjection{ProjectionType: "ALL"}},
+			"gA": {KeySchema: &pb.DynamoKeySchema{HashKey: "alpha"}, Projection: &pb.DynamoGSIProjection{ProjectionType: "ALL"}},
+			"gM": {KeySchema: &pb.DynamoKeySchema{HashKey: "mu"}, Projection: &pb.DynamoGSIProjection{ProjectionType: "ALL"}},
+		},
+		Generation: 1,
+	}
+	// Run schemaToPublic many times — Go's randomised map order
+	// would otherwise produce different array orders across calls.
+	want := schemaToPublic(schema)
+	for i := 0; i < 32; i++ {
+		got := schemaToPublic(schema)
+		if !attributeDefinitionsEqual(got.AttributeDefinitions, want.AttributeDefinitions) {
+			t.Fatalf("attribute_definitions order differs across calls: %+v vs %+v",
+				got.AttributeDefinitions, want.AttributeDefinitions)
+		}
+		if !gsiOrderEqual(got.GlobalSecondaryIndexes, want.GlobalSecondaryIndexes) {
+			t.Fatalf("global_secondary_indexes order differs across calls: %+v vs %+v",
+				got.GlobalSecondaryIndexes, want.GlobalSecondaryIndexes)
+		}
+	}
+	// Also assert the order itself is the documented sort-by-name.
+	wantAttrOrder := []string{"alpha", "id", "mu", "zeta"}
+	for i, ad := range want.AttributeDefinitions {
+		if ad.Name != wantAttrOrder[i] {
+			t.Fatalf("attribute_definitions[%d].Name = %q want %q", i, ad.Name, wantAttrOrder[i])
+		}
+	}
+	wantGSIOrder := []string{"gA", "gM", "gZ"}
+	for i, g := range want.GlobalSecondaryIndexes {
+		if g.Name != wantGSIOrder[i] {
+			t.Fatalf("global_secondary_indexes[%d].Name = %q want %q", i, g.Name, wantGSIOrder[i])
+		}
+	}
+}
+
+func attributeDefinitionsEqual(a, b []publicAttributeDefinition) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func gsiOrderEqual(a, b []publicGSI) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i].Name != b[i].Name {
+			return false
+		}
+	}
+	return true
+}
+
 func TestDDB_NewGenerationWinsOverMigrationSourceForSameKey(t *testing.T) {
 	t.Parallel()
 	enc, root := newDDBEncoder(t)