Merge remote-tracking branch 'origin/feat/pullsync-soc-convergence' into feat/pullsync-soc-convergence

Author: sbackend123

pkg/storer/internal/reserve/convergence_test.go

@@ -0,0 +1,363 @@
+// Copyright 2026 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package reserve_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/ethersphere/bee/v2/pkg/cac"
+	"github.com/ethersphere/bee/v2/pkg/crypto"
+	"github.com/ethersphere/bee/v2/pkg/log"
+	postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing"
+	"github.com/ethersphere/bee/v2/pkg/soc"
+	"github.com/ethersphere/bee/v2/pkg/storage"
+	"github.com/ethersphere/bee/v2/pkg/storer/internal"
+	"github.com/ethersphere/bee/v2/pkg/storer/internal/reserve"
+	"github.com/ethersphere/bee/v2/pkg/storer/internal/transaction"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+	kademlia "github.com/ethersphere/bee/v2/pkg/topology/mock"
+)
+
+// This file is a reusable conflict-testing harness for reserve.Put.
+//
+// The property under test is ARRIVAL-ORDER CONVERGENCE: for any set of
+// conflicting chunks, every permutation of arrivals must leave the reserve in
+// the same final state. Different nodes receive the same chunks in different
+// orders; any order-dependent outcome means neighborhoods that can never
+// agree, which breaks the redistribution game. The fingerprint deliberately
+// ignores bin IDs (they are order-dependent by design) and compares which
+// entries exist and which content each entry serves. Index integrity (the
+// chunk sum index in lockstep with the chunk bin index and the chunkstore
+// payloads) is asserted on every permutation as a side effect.
+//
+// To reuse against a reserve.Put refactor: keep the corner-case table, run
+// `go test ./pkg/storer/internal/reserve/ -run TestPutOrderConvergence -v`.
+// Cases documenting currently unresolved outcomes are marked in the table.
+
+// benignPutErr reports errors that are legitimate per-chunk outcomes of Put
+// rather than failures: losing a tie-break or carrying an older timestamp.
+func benignPutErr(err error) bool {
+	return errors.Is(err, storage.ErrOverwriteNewerChunk) ||
+		errors.Is(err, storage.ErrDivergentChunkRejected)
+}
+
+// reserveFingerprint canonicalizes the reserve state: one line per reserve
+// entry (batch, bin, address, stamp hash and the keccak of the payload it
+// currently serves) plus the full chunk sum index. While fingerprinting it
+// asserts the cross-index invariants.
+func reserveFingerprint(t *testing.T, st transaction.Storage) string {
+	t.Helper()
+	ctx := context.Background()
+
+	var lines []string
+
+	err := st.IndexStore().Iterate(
+		storage.Query{Factory: func() storage.Item { return &reserve.BatchRadiusItem{} }},
+		func(res storage.Result) (bool, error) {
+			item := res.Entry.(*reserve.BatchRadiusItem)
+			ch, err := st.ChunkStore().Get(ctx, item.Address)
+			if err != nil {
+				return false, fmt.Errorf("entry %s has no payload: %w", item.Address, err)
+			}
+			h := swarm.NewHasher()
+			_, _ = h.Write(ch.Data())
+			lines = append(lines, fmt.Sprintf("entry batch=%x addr=%x stamphash=%x content=%x",
+				item.BatchID[:8], item.Address.Bytes()[:8], item.StampHash[:8], h.Sum(nil)[:8]))
+			return false, nil
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// chunk bin index: sums must match the payload actually served
+	binSums := make(map[string]struct{})
+	err = st.IndexStore().Iterate(
+		storage.Query{Factory: func() storage.Item { return &reserve.ChunkBinItem{} }},
+		func(res storage.Result) (bool, error) {
+			cbi := res.Entry.(*reserve.ChunkBinItem)
+			ch, err := st.ChunkStore().Get(ctx, cbi.Address)
+			if err != nil {
+				return false, fmt.Errorf("bin entry %s has no payload: %w", cbi.Address, err)
+			}
+			want, err := storage.ChunkSumFromParts(cbi.BatchID, cbi.StampHash, ch)
+			if err != nil {
+				return false, err
+			}
+			if !bytes.Equal(cbi.Sum, want) {
+				return false, fmt.Errorf("stale sum on entry %s", cbi.Address)
+			}
+			binSums[cbi.Address.ByteString()+string(cbi.Sum)] = struct{}{}
+			lines = append(lines, fmt.Sprintf("sum addr=%x sum=%x", cbi.Address.Bytes()[:8], cbi.Sum[:8]))
+			return false, nil
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// chunk sum index: exactly the live (address, sum) set
+	err = st.IndexStore().Iterate(
+		storage.Query{
+			Factory:      func() storage.Item { return &reserve.ChunkSumItem{} },
+			ItemProperty: storage.QueryItemID,
+		},
+		func(res storage.Result) (bool, error) {
+			if _, ok := binSums[res.ID]; !ok {
+				return false, errors.New("orphaned chunk sum entry")
+			}
+			delete(binSums, res.ID)
+			return false, nil
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(binSums) != 0 {
+		t.Fatal("live entries missing from the chunk sum index")
+	}
+
+	sort.Strings(lines)
+	return strings.Join(lines, "\n")
+}
+
+func permutations(n int) [][]int {
+	if n == 1 {
+		return [][]int{{0}}
+	}
+	var out [][]int
+	for _, sub := range permutations(n - 1) {
+		for pos := 0; pos <= len(sub); pos++ {
+			p := make([]int, 0, n)
+			p = append(p, sub[:pos]...)
+			p = append(p, n-1)
+			p = append(p, sub[pos:]...)
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+// runOrder applies the chunks to a fresh reserve in the given order and
+// returns the state fingerprint.
+func runOrder(t *testing.T, chunks []swarm.Chunk, order []int) string {
+	t.Helper()
+
+	baseAddr := swarm.NewAddress(make([]byte, swarm.HashSize)) // fixed base: bins irrelevant here
+	st := internal.NewInmemStorage()
+	r, err := reserve.New(baseAddr, st, 0, kademlia.NewTopologyDriver(), log.Noop)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, i := range order {
+		if err := r.Put(context.Background(), chunks[i]); err != nil && !benignPutErr(err) {
+			t.Fatalf("order %v chunk %d: %v", order, i, err)
+		}
+	}
+	return reserveFingerprint(t, st)
+}
+
+// assertOrderConvergence checks every permutation. For cases marked
+// unresolved the divergence is reported without failing, so the harness
+// documents the open holes while keeping the suite green; set
+// RESERVE_STRICT_CONVERGENCE=1 to turn them into failures (useful while
+// refactoring reserve.Put toward full order independence). An unresolved case
+// that starts converging fails loudly so the marker gets removed.
+func assertOrderConvergence(t *testing.T, chunks []swarm.Chunk, unresolved bool) {
+	t.Helper()
+
+	strict := os.Getenv("RESERVE_STRICT_CONVERGENCE") != ""
+	perms := permutations(len(chunks))
+	first := runOrder(t, chunks, perms[0])
+	for _, p := range perms[1:] {
+		fp := runOrder(t, chunks, p)
+		if fp != first {
+			msg := fmt.Sprintf("order-dependent outcome:\norder %v ends with:\n%s\n\norder %v ends with:\n%s",
+				perms[0], first, p, fp)
+			if unresolved && !strict {
+				t.Logf("KNOWN UNRESOLVED (not failing, set RESERVE_STRICT_CONVERGENCE=1 to enforce):\n%s", msg)
+			} else {
+				t.Error(msg)
+			}
+			return
+		}
+	}
+	if unresolved {
+		t.Error("case marked unresolved now converges: remove the unresolved marker")
+	}
+}
+
+func newTestSOC(t *testing.T, signer crypto.Signer, id, payload []byte) swarm.Chunk {
+	t.Helper()
+	inner, err := cac.New(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ch, err := soc.New(id, inner).Sign(signer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ch
+}
+
+func newTestCAC(t *testing.T, payload []byte) swarm.Chunk {
+	t.Helper()
+	ch, err := cac.New(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ch
+}
+
+// TestPutOrderConvergence drives conflicting chunk sets through every arrival
+// order and requires an identical final state.
+func TestPutOrderConvergence(t *testing.T) {
+	t.Parallel()
+
+	signer := getSigner(t)
+	batchA := postagetesting.MustNewBatch()
+	batchB := postagetesting.MustNewBatch()
+	id1 := make([]byte, swarm.HashSize)
+	id2 := bytes.Repeat([]byte{1}, swarm.HashSize)
+
+	for _, tc := range []struct {
+		name       string
+		unresolved bool
+		chunks     func(t *testing.T) []swarm.Chunk
+	}{
+		{
+			// Sofia's rule 4 core case: the tie-break must make this converge.
+			name: "cac vs cac, same slot, equal timestamp",
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestCAC(t, []byte("cac payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+					newTestCAC(t, []byte("cac payload two")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+				}
+			},
+		},
+		{
+			// Phase 2 core case: same SOC address, byte-identical stamp,
+			// different payloads; resolveDivergence must make this converge.
+			name: "divergent socs, identical stamp",
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				stamp := postagetesting.MustNewFields(batchA.ID, 0, 7)
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(stamp),
+					newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(stamp),
+				}
+			},
+		},
+		{
+			// Newer timestamps must win regardless of order or type.
+			name: "soc update, increasing timestamps",
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc v1")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 1)),
+					newTestSOC(t, signer, id1, []byte("soc v2")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 2)),
+				}
+			},
+		},
+		{
+			// UNRESOLVED on this branch: same SOC address, same slot and
+			// timestamp, but separately stamped (distinct signatures, hence
+			// distinct stamp hashes). Bypasses resolveDivergence (stamp hashes
+			// differ) and the CAC tie-break (wrong type): last write wins.
+			name:       "divergent socs, equal timestamp, distinct stamps",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+					newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+				}
+			},
+		},
+		{
+			// UNRESOLVED on this branch: different SOC addresses in the same
+			// slot at the same timestamp. Not content addressed, so the
+			// tie-break is skipped: last write wins.
+			name:       "soc vs soc, different addresses, same slot, equal timestamp",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+					newTestSOC(t, signer, id2, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+				}
+			},
+		},
+		{
+			// UNRESOLVED on this branch: mixed types in the same slot at the
+			// same timestamp. The tie-break fires only when the INCOMING chunk
+			// is content addressed, so the two directions disagree whenever
+			// the CAC has the lower address.
+			name:       "cac vs soc, same slot, equal timestamp, cac address lower",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				socCh := newTestSOC(t, signer, id1, []byte("soc payload"))
+				// search a payload whose CAC address sorts below the SOC's
+				for i := range 64 {
+					cacCh := newTestCAC(t, fmt.Appendf(nil, "cac payload %d", i))
+					if bytes.Compare(cacCh.Address().Bytes(), socCh.Address().Bytes()) < 0 {
+						return []swarm.Chunk{
+							socCh.WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+							cacCh.WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+						}
+					}
+				}
+				t.Fatal("no lower cac address found")
+				return nil
+			},
+		},
+		{
+			// UNRESOLVED on this branch: byte-identical CAC re-stamped in the
+			// same slot at the same timestamp with a different signature. The
+			// entries swap stamp hashes depending on order, so peers holding
+			// different stampings keep exchanging and replacing forever.
+			name:       "identical cac, same slot, equal timestamp, distinct stamps",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				payload := []byte("identical cac payload")
+				return []swarm.Chunk{
+					newTestCAC(t, payload).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+					newTestCAC(t, payload).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 7)),
+				}
+			},
+		},
+		{
+			// Three-way conflict across batches: the batch B entry must end
+			// serving whatever payload the batch A conflict settles on, with
+			// its sum refreshed accordingly.
+			name:       "divergent socs with a sibling entry under another batch",
+			unresolved: true,
+			chunks: func(t *testing.T) []swarm.Chunk {
+				t.Helper()
+				stamp := postagetesting.MustNewFields(batchA.ID, 0, 7)
+				return []swarm.Chunk{
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(stamp),
+					newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(stamp),
+					newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchB.ID, 0, 7)),
+				}
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assertOrderConvergence(t, tc.chunks(t), tc.unresolved)
+		})
+	}
+}