Skip to content

Commit 9318f24

Browse files
committed
Merge remote-tracking branch 'origin/feat/pullsync-soc-convergence' into feat/pullsync-soc-convergence
# Conflicts: # pkg/storer/internal/reserve/reserve.go
2 parents 304601f + 49584b0 commit 9318f24

10 files changed

Lines changed: 875 additions & 344 deletions

File tree

.github/workflows/beekeeper.yml

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ env:
1919
SETUP_CONTRACT_IMAGE: "ethersphere/bee-localchain"
2020
SETUP_CONTRACT_IMAGE_TAG: "0.9.4"
2121
BEELOCAL_BRANCH: "main"
22-
BEEKEEPER_BRANCH: "master"
22+
BEEKEEPER_BRANCH: "feat/pullsync-chuns-convergence"
2323
BEEKEEPER_METRICS_ENABLED: false
2424
REACHABILITY_OVERRIDE_PUBLIC: true
2525
BATCHFACTOR_OVERRIDE_PUBLIC: 2
@@ -153,10 +153,10 @@ jobs:
153153
run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-gsoc
154154
- name: Test pushsync (chunks)
155155
id: pushsync-chunks-1
156-
run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-pushsync-chunks
156+
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'
157157
- name: Test pushsync (light mode chunks)
158158
id: pushsync-chunks-2
159-
run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-pushsync-light-chunks
159+
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'
160160
- name: Test retrieval
161161
id: retrieval
162162
run: timeout ${TIMEOUT} beekeeper check --cluster-name local-dns --checks=ci-retrieval

go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ require (
1111
github.com/ethereum/go-ethereum v1.17.3
1212
github.com/ethersphere/batch-archive v0.0.8
1313
github.com/ethersphere/go-price-oracle-abi v0.6.9
14-
github.com/ethersphere/go-storage-incentives-abi v0.9.4
14+
github.com/ethersphere/go-storage-incentives-abi v0.9.3-rc4
1515
github.com/ethersphere/go-sw3-abi v0.6.9
1616
github.com/ethersphere/langos v1.0.0
1717
github.com/go-playground/validator/v10 v10.19.0

go.sum

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -250,8 +250,8 @@ github.com/ethersphere/batch-archive v0.0.8 h1:Y6ipqJfcjLbOn+2Rn5tMrOvrMH7pzF0Yh
250250
github.com/ethersphere/batch-archive v0.0.8/go.mod h1:41BPb192NoK9CYjNB8BAE1J2MtiI/5aq0Wtas5O7A7Q=
251251
github.com/ethersphere/go-price-oracle-abi v0.6.9 h1:bseen6he3PZv5GHOm+KD6s4awaFmVSD9LFx+HpB6rCU=
252252
github.com/ethersphere/go-price-oracle-abi v0.6.9/go.mod h1:sI/Qj4/zJ23/b1enzwMMv0/hLTpPNVNacEwCWjo6yBk=
253-
github.com/ethersphere/go-storage-incentives-abi v0.9.4 h1:mSIWXQXg5OQmH10QvXMV5w0vbSibFMaRlBL37gPLTM0=
254-
github.com/ethersphere/go-storage-incentives-abi v0.9.4/go.mod h1:SXvJVtM4sEsaSKD0jc1ClpDLw8ErPoROZDme4Wrc/Nc=
253+
github.com/ethersphere/go-storage-incentives-abi v0.9.3-rc4 h1:YK9FpiQz29ctU5V46CuwMt+4X5Xn8FTBwy6E2v/ix8s=
254+
github.com/ethersphere/go-storage-incentives-abi v0.9.3-rc4/go.mod h1:SXvJVtM4sEsaSKD0jc1ClpDLw8ErPoROZDme4Wrc/Nc=
255255
github.com/ethersphere/go-sw3-abi v0.6.9 h1:TnWLnYkWE5UvC17mQBdUmdkzhPhO8GcqvWy4wvd1QJQ=
256256
github.com/ethersphere/go-sw3-abi v0.6.9/go.mod h1:BmpsvJ8idQZdYEtWnvxA8POYQ8Rl/NhyCdF0zLMOOJU=
257257
github.com/ethersphere/langos v1.0.0 h1:NBtNKzXTTRSue95uOlzPN4py7Aofs0xWPzyj4AI1Vcc=

pkg/pullsync/pullsync.go

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -345,6 +345,7 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
345345
}
346346

347347
wantChunkID := addr.ByteString() + string(sum)
348+
348349
if _, ok := wantChunks[wantChunkID]; !ok {
349350
s.logger.Debug("want chunks", "error", ErrUnsolicitedChunk, "peer_address", peer, "chunk_address", addr)
350351
chunkErr = errors.Join(chunkErr, ErrUnsolicitedChunk)
@@ -398,7 +399,6 @@ func (s *Syncer) Sync(ctx context.Context, peer swarm.Address, bin uint8, start
398399
// tie-break. The neighborhood converges on the stored chunk, so
399400
// this is an expected outcome rather than a sync error.
400401
if errors.Is(err, storage.ErrDivergentChunkRejected) {
401-
s.logger.Debug("divergent chunk rejected", "error", err, "peer_address", peer, "chunk", c)
402402
s.metrics.DivergentRejected.Inc()
403403
continue
404404
}
@@ -426,6 +426,7 @@ func (s *Syncer) makeOffer(ctx context.Context, rn pb.Get) (*pb.Offer, []*storer
426426
o.Chunks = make([]*pb.Chunk, 0, len(bincs))
427427
for _, v := range bincs {
428428
o.Chunks = append(o.Chunks, &pb.Chunk{Address: v.Address.Bytes(), Sum: v.Sum})
429+
429430
}
430431
return o, bincs, nil
431432
}
@@ -465,7 +466,7 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
465466
break LOOP // The stream has been closed.
466467
}
467468

468-
chs = append(chs, &storer.BinC{Address: c.Address, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
469+
chs = append(chs, &storer.BinC{Address: c.Address, BinID: c.BinID, BatchID: c.BatchID, StampHash: c.StampHash, Sum: c.Sum})
469470
if c.BinID > topmost {
470471
topmost = c.BinID
471472
}

pkg/storer/internal/reserve/convergence_test.go

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import (
1717
"github.com/ethersphere/bee/v2/pkg/cac"
1818
"github.com/ethersphere/bee/v2/pkg/crypto"
1919
"github.com/ethersphere/bee/v2/pkg/log"
20+
"github.com/ethersphere/bee/v2/pkg/postage"
2021
postagetesting "github.com/ethersphere/bee/v2/pkg/postage/testing"
2122
"github.com/ethersphere/bee/v2/pkg/soc"
2223
"github.com/ethersphere/bee/v2/pkg/storage"
@@ -330,6 +331,37 @@ func TestPutOrderConvergence(t *testing.T) {
330331
}
331332
},
332333
},
334+
{
335+
// Same SOC address, same batch, equal timestamp, different stamp
336+
// indices. putSOC treats this as a new stamp entry and replaces the
337+
// shared payload unconditionally (last-write wins). Desired: settle
338+
// on the lexicographically lower stamp hash like the same-slot path.
339+
name: "divergent socs, equal timestamp, distinct stamp indices",
340+
unresolved: true,
341+
chunks: func(t *testing.T) []swarm.Chunk {
342+
t.Helper()
343+
return []swarm.Chunk{
344+
newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 5)),
345+
newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchA.ID, 1, 5)),
346+
}
347+
},
348+
},
349+
{
350+
// Same SOC address under two batches at the same timestamp.
351+
// putSOC currently replaces the shared payload on the second stamp
352+
// unconditionally (last-write wins), so arrival order decides the
353+
// payload. Desired: settle on the lexicographically lower stamp
354+
// hash, matching the same-slot equal-timestamp path.
355+
name: "divergent socs, equal timestamp, distinct batches",
356+
unresolved: true,
357+
chunks: func(t *testing.T) []swarm.Chunk {
358+
t.Helper()
359+
return []swarm.Chunk{
360+
newTestSOC(t, signer, id1, []byte("soc payload one")).WithStamp(postagetesting.MustNewFields(batchA.ID, 0, 5)),
361+
newTestSOC(t, signer, id1, []byte("soc payload two")).WithStamp(postagetesting.MustNewFields(batchB.ID, 0, 5)),
362+
}
363+
},
364+
},
333365
{
334366
// Three-way conflict across batches: the batch B entry must end
335367
// serving whatever payload the batch A conflict settles on, with
@@ -353,3 +385,87 @@ func TestPutOrderConvergence(t *testing.T) {
353385
})
354386
}
355387
}
388+
389+
func TestSOCMultiStampDivergenceCornerCase(t *testing.T) {
390+
t.Parallel()
391+
392+
baseAddr := swarm.RandAddress(t)
393+
ts := internal.NewInmemStorage()
394+
r, err := reserve.New(baseAddr, ts, 0, kademlia.NewTopologyDriver(), log.Noop)
395+
if err != nil {
396+
t.Fatal(err)
397+
}
398+
399+
privKey, err := crypto.GenerateSecp256k1Key()
400+
if err != nil {
401+
t.Fatal(err)
402+
}
403+
signer := crypto.NewDefaultSigner(privKey)
404+
idBytes := make([]byte, 32)
405+
406+
chCAC1, err := cac.New([]byte("payload-1-alpha"))
407+
if err != nil {
408+
t.Fatal(err)
409+
}
410+
chCAC2, err := cac.New([]byte("payload-2-beta"))
411+
if err != nil {
412+
t.Fatal(err)
413+
}
414+
if bytes.Compare(chCAC1.Address().Bytes(), chCAC2.Address().Bytes()) > 0 {
415+
chCAC1, chCAC2 = chCAC2, chCAC1
416+
}
417+
418+
soc1, err := soc.New(idBytes, chCAC1).Sign(signer)
419+
if err != nil {
420+
t.Fatal(err)
421+
}
422+
soc2, err := soc.New(idBytes, chCAC2).Sign(signer)
423+
if err != nil {
424+
t.Fatal(err)
425+
}
426+
427+
var stampA, stampB *postage.Stamp
428+
for {
429+
batchA := postagetesting.MustNewBatch()
430+
batchB := postagetesting.MustNewBatch()
431+
stA := postagetesting.MustNewFields(batchA.ID, 0, 1000)
432+
stB := postagetesting.MustNewFields(batchB.ID, 0, 1000)
433+
shA, _ := stA.Hash()
434+
shB, _ := stB.Hash()
435+
if bytes.Compare(shB, shA) < 0 {
436+
stampA, stampB = stA, stB
437+
break
438+
}
439+
}
440+
441+
ctx := context.Background()
442+
443+
// 1. Put Stamp A + Payload P1 (soc1)
444+
err = r.Put(ctx, soc1.WithStamp(stampA))
445+
if err != nil {
446+
t.Fatalf("put soc1 stampA failed: %v", err)
447+
}
448+
449+
// 2. Put Stamp B + Payload P2 (soc2) under same timestamp.
450+
// Since stampHashB < stampHashA, Stamp B wins over Stamp A.
451+
err = r.Put(ctx, soc2.WithStamp(stampB))
452+
if err != nil {
453+
t.Fatalf("put soc2 stampB failed: %v", err)
454+
}
455+
456+
// 3. Re-offer Stamp A + Payload P1 (soc1).
457+
// Stamp A lost to Stamp B at timestamp 1000. Re-offering Stamp A + P1 MUST NOT restore P1!
458+
err = r.Put(ctx, soc1.WithStamp(stampA))
459+
if err == nil {
460+
t.Fatalf("expected ErrDivergentChunkRejected when re-offering weaker stampA, got nil")
461+
}
462+
463+
// Verify that active chunk in ChunkStore STILL has Payload P2 (soc2)
464+
finalCh, err := ts.ChunkStore().Get(ctx, soc1.Address())
465+
if err != nil {
466+
t.Fatalf("get final chunk failed: %v", err)
467+
}
468+
if !bytes.Equal(finalCh.Data(), soc2.Data()) {
469+
t.Fatalf("re-offered Stamp A restored payload P1 over Stamp B's winning payload P2!")
470+
}
471+
}

0 commit comments

Comments
 (0)