feat: chunks convergence (#5557)

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: Calin Martinconi <martinconic@gmail.com>
Co-authored-by: Akrem Chabchoub <121046693+akrem-chabchoub@users.noreply.github.com>
Co-authored-by: Not Darko <93942788+darkobas2@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Ljubiša Gačević <35105035+gacevicljubisa@users.noreply.github.com>
Co-authored-by: Janoš Guljaš <janos@users.noreply.github.com>
Co-authored-by: acud <12988138+acud@users.noreply.github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Author: 9 people

.github/workflows/beekeeper.yml

@@ -19,7 +19,7 @@ env:
   SETUP_CONTRACT_IMAGE: "ethersphere/bee-localchain"
   SETUP_CONTRACT_IMAGE_TAG: "0.9.4"
   BEELOCAL_BRANCH: "main"
-  BEEKEEPER_BRANCH: "master"
+  BEEKEEPER_BRANCH: "feat/pullsync-chuns-convergence"
   BEEKEEPER_METRICS_ENABLED: false
   REACHABILITY_OVERRIDE_PUBLIC: true
   BATCHFACTOR_OVERRIDE_PUBLIC: 2
@@ -151,12 +151,16 @@ jobs:
       - name: Test gsoc
         id: gsoc
         run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-gsoc
+      - name: Test socmatrix
+        id: socmatrix
+        # Check allows up to 60m; wall clock is typically ~20–30m (16 scenarios + sync-wait).
+        run: timeout 60m beekeeper check --cluster-name local-dns --checks=ci-socmatrix
       - name: Test pushsync (chunks)
         id: pushsync-chunks-1
-        run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-pushsync-chunks
+        run: timeout ${TIMEOUT} bash -c 'until beekeeper check --cluster-name local-dns --checks=ci-pushsync-chunks; do echo "waiting for pushsync-chunks..."; sleep .3; done'
       - name: Test pushsync (light mode chunks)
         id: pushsync-chunks-2
-        run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-pushsync-light-chunks
+        run: timeout ${TIMEOUT} bash -c 'until beekeeper check --cluster-name local-dns --checks=ci-pushsync-light-chunks; do echo "waiting for pushsync-light-chunks..."; sleep .3; done'
       - name: Test retrieval
         id: retrieval
         run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-retrieval
@@ -198,6 +202,7 @@ jobs:
           if ${{ steps.pss.outcome=='failure' }}; then FAILED=pss; fi
           if ${{ steps.soc.outcome=='failure' }}; then FAILED=soc; fi
           if ${{ steps.gsoc.outcome=='failure' }}; then FAILED=gsoc; fi
+          if ${{ steps.socmatrix.outcome=='failure' }}; then FAILED=socmatrix; fi
           if ${{ steps.pushsync-chunks-1.outcome=='failure' }}; then FAILED=pushsync-chunks-1; fi
           if ${{ steps.pushsync-chunks-2.outcome=='failure' }}; then FAILED=pushsync-chunks-2; fi
           if ${{ steps.retrieval.outcome=='failure' }}; then FAILED=retrieval; fi

pkg/pullsync/metrics.go

@@ -15,6 +15,7 @@ type metrics struct {
 	MissingChunks        prometheus.Counter     // number of reserve get errs
 	ReceivedZeroAddress  prometheus.Counter     // number of delivered chunks with invalid address
 	ReceivedInvalidChunk prometheus.Counter     // number of delivered chunks with invalid address
+	DivergentRejected    prometheus.Counter     // number of delivered chunks that lost the divergence tie-break
 	Delivered            prometheus.Counter     // number of chunk deliveries
 	SentOffered          prometheus.Counter     // number of chunks offered
 	SentWanted           prometheus.Counter     // number of chunks wanted
@@ -57,6 +58,12 @@ func newMetrics() metrics {
 			Name:      "received_invalid_chunks",
 			Help:      "Total invalid chunks delivered.",
 		}),
+		DivergentRejected: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "divergent_rejected",
+			Help:      "Total delivered chunks discarded for losing the divergence tie-break.",
+		}),
 		Delivered: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,

pkg/pullsync/pullsync.go

@@ -394,6 +394,13 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
 					chunkErr = errors.Join(chunkErr, err)
 					continue
 				}
+				// the chunk diverged from the one already stored and lost the
+				// tie-break. The neighborhood converges on the stored chunk, so
+				// this is an expected outcome rather than a sync error.
+				if errors.Is(err, storage.ErrDivergentChunkRejected) {
+					s.metrics.DivergentRejected.Inc()
+					continue
+				}
 				return 0, 0, errors.Join(chunkErr, err)
 			}
 			chunksPut++
@@ -457,7 +464,7 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
 					break LOOP // The stream has been closed.
 				}
 
-				chs = append(chs, &storer.BinC{Address: c.Address, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
+				chs = append(chs, &storer.BinC{Address: c.Address, BinID: c.BinID, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
 				if c.BinID > topmost {
 					topmost = c.BinID
 				}

pkg/storage/storage.go

@@ -5,6 +5,7 @@
 package storage
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -19,6 +20,12 @@ import (
 var (
 	ErrOverwriteNewerChunk = errors.New("overwriting chunk with newer timestamp")
 	ErrUnknownChunkType    = errors.New("unknown chunk type")
+
+	// ErrDivergentChunkRejected is returned when a chunk that diverges from an
+	// already stored one at the same address, batch and stamp loses the
+	// deterministic tie-break and is therefore not stored. It is not a failure:
+	// the node already holds the chunk the whole neighborhood converges on.
+	ErrDivergentChunkRejected = errors.New("divergent chunk rejected by tie-break")
 )
 
 // Result represents the item returned by the read operation, which returns
@@ -336,6 +343,39 @@ func ChunkSumFromParts(batchID, stampHash []byte, ch swarm.Chunk) ([]byte, error
 	return h.Sum(nil)[:ChunkSumSize], nil
 }
 
+// DivergentSocChunkWins reports whether the incoming chunk should replace the
+// stored one when the two share an address, batch and stamp but wrap different
+// content. Both chunks must be single owner chunks; a content addressed chunk
+// cannot diverge, since its address is the hash of its own payload.
+//
+// The winner is the chunk wrapping the lexicographically lower CAC address.
+// The rule depends on nothing but the two payloads, so every node in the
+// neighborhood converges on the same chunk regardless of the order in which
+// they arrive.
+func DivergentSocChunkWins(stored, incoming swarm.Chunk) (bool, error) {
+	storedAddr, err := wrappedAddress(stored)
+	if err != nil {
+		return false, fmt.Errorf("stored chunk: %w", err)
+	}
+	incomingAddr, err := wrappedAddress(incoming)
+	if err != nil {
+		return false, fmt.Errorf("incoming chunk: %w", err)
+	}
+	return bytes.Compare(incomingAddr.Bytes(), storedAddr.Bytes()) < 0, nil
+}
+
+// wrappedAddress returns the address of the CAC wrapped by a single owner chunk.
+func wrappedAddress(ch swarm.Chunk) (swarm.Address, error) {
+	if !soc.Valid(ch) {
+		return swarm.ZeroAddress, fmt.Errorf("%w: not a single owner chunk", ErrUnknownChunkType)
+	}
+	s, err := soc.FromChunk(ch)
+	if err != nil {
+		return swarm.ZeroAddress, fmt.Errorf("soc from chunk: %w", err)
+	}
+	return s.WrappedChunk().Address(), nil
+}
+
 // IdentityAddress returns the internally used address for the chunk
 // since the single owner chunk address is not a unique identifier for the chunk,
 // but hashing the soc address and the wrapped chunk address is.

pkg/storage/storage_test.go

@@ -6,6 +6,7 @@ package storage_test
 import (
 	"bytes"
 	"encoding/hex"
+	"errors"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -69,7 +70,7 @@ func TestIdentityAddress(t *testing.T) {
 		data := []byte("data")
 		cacChunk, err := cac.New(data)
 		if err != nil {
-			t.Fatalf("failed to create content addressed chunk: %v", err)
+			t.Fatalf("create content addressed chunk: %v", err)
 		}
 
 		// Call IdentityAddress with the CAC
@@ -234,3 +235,88 @@ func FuzzChunkSum(f *testing.F) {
 		}
 	})
 }
+
+func TestDivergentSocChunkWins(t *testing.T) {
+	t.Parallel()
+
+	privKey, err := crypto.GenerateSecp256k1Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	signer := crypto.NewDefaultSigner(privKey)
+	id := make([]byte, swarm.HashSize)
+
+	newSOC := func(data string) swarm.Chunk {
+		t.Helper()
+		inner, err := cac.New([]byte(data))
+		if err != nil {
+			t.Fatal(err)
+		}
+		ch, err := soc.New(id, inner).Sign(signer)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return ch.WithStamp(postagetesting.MustNewStamp())
+	}
+
+	lower, higher := newSOC("content-one"), newSOC("content-two")
+	lowerInner, err := soc.UnwrapCAC(lower)
+	if err != nil {
+		t.Fatal(err)
+	}
+	higherInner, err := soc.UnwrapCAC(higher)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Compare(lowerInner.Address().Bytes(), higherInner.Address().Bytes()) > 0 {
+		lower, higher = higher, lower
+	}
+
+	t.Run("lower wrapped address wins", func(t *testing.T) {
+		t.Parallel()
+
+		wins, err := storage.DivergentSocChunkWins(higher, lower)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !wins {
+			t.Fatal("expected the chunk wrapping the lower cac address to win")
+		}
+	})
+
+	t.Run("tie-break is antisymmetric", func(t *testing.T) {
+		t.Parallel()
+
+		wins, err := storage.DivergentSocChunkWins(lower, higher)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if wins {
+			t.Fatal("expected the chunk wrapping the higher cac address to lose")
+		}
+	})
+
+	t.Run("a chunk does not displace itself", func(t *testing.T) {
+		t.Parallel()
+
+		wins, err := storage.DivergentSocChunkWins(lower, lower)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if wins {
+			t.Fatal("expected an identical chunk not to win")
+		}
+	})
+
+	t.Run("content addressed chunks cannot diverge", func(t *testing.T) {
+		t.Parallel()
+
+		cac := testingc.GenerateTestRandomChunk()
+		if _, err := storage.DivergentSocChunkWins(cac, lower); !errors.Is(err, storage.ErrUnknownChunkType) {
+			t.Fatalf("expected ErrUnknownChunkType, got %v", err)
+		}
+		if _, err := storage.DivergentSocChunkWins(lower, cac); !errors.Is(err, storage.ErrUnknownChunkType) {
+			t.Fatalf("expected ErrUnknownChunkType, got %v", err)
+		}
+	})
+}