fix(storer): enforce timestamp and stamp-hash precedence in resolveDivergence

Author: martinconic

pkg/storer/internal/reserve/convergence_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/cac"
 	"github.com/ethersphere/bee/v2/pkg/crypto"
 	"github.com/ethersphere/bee/v2/pkg/log"
+	"github.com/ethersphere/bee/v2/pkg/postage"
 	postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing"
 	"github.com/ethersphere/bee/v2/pkg/soc"
 	"github.com/ethersphere/bee/v2/pkg/storage"
@@ -384,3 +385,87 @@ func TestPutOrderConvergence(t *testing.T) {
 		})
 	}
 }
+
+func TestSOCMultiStampDivergenceCornerCase(t *testing.T) {
+	t.Parallel()
+
+	baseAddr := swarm.RandAddress(t)
+	ts := internal.NewInmemStorage()
+	r, err := reserve.New(baseAddr, ts, 0, kademlia.NewTopologyDriver(), log.Noop)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	privKey, err := crypto.GenerateSecp256k1Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	signer := crypto.NewDefaultSigner(privKey)
+	idBytes := make([]byte, 32)
+
+	chCAC1, err := cac.New([]byte("payload-1-alpha"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	chCAC2, err := cac.New([]byte("payload-2-beta"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Compare(chCAC1.Address().Bytes(), chCAC2.Address().Bytes()) > 0 {
+		chCAC1, chCAC2 = chCAC2, chCAC1
+	}
+
+	soc1, err := soc.New(idBytes, chCAC1).Sign(signer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	soc2, err := soc.New(idBytes, chCAC2).Sign(signer)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var stampA, stampB *postage.Stamp
+	for {
+		batchA := postagetesting.MustNewBatch()
+		batchB := postagetesting.MustNewBatch()
+		stA := postagetesting.MustNewFields(batchA.ID, 0, 1000)
+		stB := postagetesting.MustNewFields(batchB.ID, 0, 1000)
+		shA, _ := stA.Hash()
+		shB, _ := stB.Hash()
+		if bytes.Compare(shB, shA) < 0 {
+			stampA, stampB = stA, stB
+			break
+		}
+	}
+
+	ctx := context.Background()
+
+	// 1. Put Stamp A + Payload P1 (soc1)
+	err = r.Put(ctx, soc1.WithStamp(stampA))
+	if err != nil {
+		t.Fatalf("put soc1 stampA failed: %v", err)
+	}
+
+	// 2. Put Stamp B + Payload P2 (soc2) under same timestamp.
+	// Since stampHashB < stampHashA, Stamp B wins over Stamp A.
+	err = r.Put(ctx, soc2.WithStamp(stampB))
+	if err != nil {
+		t.Fatalf("put soc2 stampB failed: %v", err)
+	}
+
+	// 3. Re-offer Stamp A + Payload P1 (soc1).
+	// Stamp A lost to Stamp B at timestamp 1000. Re-offering Stamp A + P1 MUST NOT restore P1!
+	err = r.Put(ctx, soc1.WithStamp(stampA))
+	if err == nil {
+		t.Fatalf("expected ErrDivergentChunkRejected when re-offering weaker stampA, got nil")
+	}
+
+	// Verify that active chunk in ChunkStore STILL has Payload P2 (soc2)
+	finalCh, err := ts.ChunkStore().Get(ctx, soc1.Address())
+	if err != nil {
+		t.Fatalf("get final chunk failed: %v", err)
+	}
+	if !bytes.Equal(finalCh.Data(), soc2.Data()) {
+		t.Fatalf("re-offered Stamp A restored payload P1 over Stamp B's winning payload P2!")
+	}
+}

pkg/storer/internal/reserve/reserve.go

@@ -371,7 +371,8 @@ func (r *Reserve) putCAC(ctx context.Context, chunk swarm.Chunk, sum, stampHash
 func (r *Reserve) resolveStampIndexCollision(
 	ctx context.Context, s transaction.Store,
 	chunk swarm.Chunk, oldStampIndex *stampindex.Item,
-	sum, stampHash []byte, bin uint8) (sameAddr bool, err error) {
+	sum, stampHash []byte, bin uint8,
+) (sameAddr bool, err error) {
 	prev := binary.BigEndian.Uint64(oldStampIndex.StampTimestamp)
 	curr := binary.BigEndian.Uint64(chunk.Stamp().Timestamp())
 	if prev > curr {
@@ -674,6 +675,31 @@ func (r *Reserve) resolveDivergence(
 			return fmt.Errorf("failed loading diverging chunk %s: %w", chunk.Address(), err)
 		}
 
+		// Verify timestamp precedence: an incoming chunk with an older timestamp
+		// can never displace a stored chunk.
+		prevTimestamp := binary.BigEndian.Uint64(stored.Stamp().Timestamp())
+		currTimestamp := binary.BigEndian.Uint64(chunk.Stamp().Timestamp())
+		if prevTimestamp > currTimestamp {
+			return fmt.Errorf("overwrite same chunk. prev %d cur %d batch %s: %w", prevTimestamp, currTimestamp, hex.EncodeToString(chunk.Stamp().BatchID()), storage.ErrOverwriteNewerChunk)
+		}
+
+		// At equal timestamp, if the stamps differ, the lower stamp hash wins.
+		if prevTimestamp == currTimestamp {
+			storedStampHash, err := stored.Stamp().Hash()
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(storedStampHash, stampHash) && bytes.Compare(storedStampHash, stampHash) < 0 {
+				r.logger.Debug(
+					"discarding diverging chunk (weaker stamp hash at equal timestamp)",
+					"address", chunk.Address(),
+					"stored_stamp_hash", hex.EncodeToString(storedStampHash),
+					"incoming_stamp_hash", hex.EncodeToString(stampHash),
+				)
+				return fmt.Errorf("diverging chunk %s lost stamp-hash tie-break: %w", chunk.Address(), storage.ErrDivergentChunkRejected)
+			}
+		}
+
 		wins, err := storage.DivergentChunkWins(stored, chunk)
 		if err != nil {
 			return fmt.Errorf("divergence tie-break for chunk %s: %w", chunk.Address(), err)
@@ -1307,6 +1333,9 @@ func (r *Reserve) IncBinID(store storage.IndexStore, bin uint8) (uint64, error)
 }
 
 func wrappedAddrHex(ch swarm.Chunk) string {
+	if ch == nil {
+		return ""
+	}
 	if !soc.Valid(ch) {
 		return ""
 	}
@@ -1328,6 +1357,7 @@ func (r *Reserve) logAddressStampState(addr swarm.Address, event string) {
 	err := chunkstamp.IterateAll(r.st.IndexStore(), reserveScope, addr, func(stamp swarm.Stamp) (bool, error) {
 		stampHash, err := stamp.Hash()
 		if err != nil {
+			// nolint:nilerr
 			return false, nil
 		}
 		item := &BatchRadiusItem{Bin: bin, BatchID: stamp.BatchID(), Address: addr, StampHash: stampHash}