Skip to content

Commit d5be79b

Browse files
committed
fix: skip primary keys with no document during head backfill
1 parent dc62a66 commit d5be79b

2 files changed

Lines changed: 87 additions & 1 deletion

File tree

internal/db/p2p/replicator.go

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -210,12 +210,17 @@ func (p *P2P) pushHeadsForAllDocs(ctx context.Context, col client.Collection, pe
210210
}
211211
}()
212212

213+
var stale int
213214
for {
214215
hasNext, err := iter.Next()
215216
if err != nil {
216217
return NewErrIterateReplicatorDocs(err)
217218
}
218219
if !hasNext {
220+
if stale > 0 {
221+
log.Info("Skipped primary keys with no document during head backfill",
222+
corelog.Int("count", stale), corelog.String("collectionID", col.CollectionID()))
223+
}
219224
return nil
220225
}
221226
primaryKey, err := keys.NewPrimaryDataStoreKey(string(iter.Key()))
@@ -231,7 +236,10 @@ func (p *P2P) pushHeadsForAllDocs(ctx context.Context, col client.Collection, pe
231236
return err
232237
}
233238
if !found {
234-
return client.ErrDocumentNotFoundOrNotAuthorized
239+
// A primary key whose document is gone. Skipping it keeps one stale key from
240+
// stopping the backfill for every document behind it.
241+
stale++
242+
continue
235243
}
236244

237245
err = p.pushHeadsForDoc(ctx, primaryKey.DocShortID, docID, col.CollectionID(), peerID)
Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
// Copyright 2026 Democratized Data Foundation
2+
//
3+
// Use of this software is governed by the Business Source License
4+
// included in the file licenses/BSL.txt.
5+
//
6+
// As of the Change Date specified in that file, in accordance with
7+
// the Business Source License, use of this software will be governed
8+
// by the Apache License, Version 2.0, included in the file
9+
// licenses/APL.txt.
10+
11+
package p2p
12+
13+
import (
14+
"context"
15+
"strconv"
16+
"testing"
17+
18+
"github.com/stretchr/testify/require"
19+
20+
"github.com/sourcenetwork/corekv"
21+
"github.com/sourcenetwork/corekv/memory"
22+
"github.com/sourcenetwork/immutable"
23+
24+
"github.com/sourcenetwork/defradb/client"
25+
"github.com/sourcenetwork/defradb/internal/datastore"
26+
"github.com/sourcenetwork/defradb/internal/db/lock"
27+
"github.com/sourcenetwork/defradb/internal/keys"
28+
)
29+
30+
// multistoreDB satisfies DB for the backfill path, which reaches no other method.
31+
type multistoreDB struct {
32+
DB
33+
stores *datastore.Multistore
34+
}
35+
36+
func (d multistoreDB) Multistore() *datastore.Multistore { return d.stores }
37+
38+
// backfillCollection carries only the identity the backfill reads off a collection.
39+
type backfillCollection struct {
40+
client.Collection
41+
collectionID string
42+
}
43+
44+
func (c backfillCollection) CollectionID() string { return c.collectionID }
45+
46+
func (c backfillCollection) Version() client.CollectionVersion {
47+
return client.CollectionVersion{CollectionID: c.collectionID}
48+
}
49+
50+
// A purge that leaves a document's primary key behind, or any other source of one, must not
51+
// stop the backfill. The iterator walks the primary prefix in key order, so returning on the
52+
// first key that will not resolve skips every document sorting after it.
53+
func TestPushHeadsForAllDocs_SkipsPrimaryKeysWithNoDocument(t *testing.T) {
54+
ctx := context.Background()
55+
rootstore := memory.NewDatastore(ctx)
56+
stores := datastore.NewMultistore(rootstore, lock.NewLockSet(), immutable.None[int]())
57+
58+
const collectionID = "bafkreicollection"
59+
const shortID = 1
60+
require.NoError(t, stores.Systemstore().Set(ctx,
61+
keys.NewCollectionID(collectionID).Bytes(), []byte(strconv.Itoa(shortID))))
62+
63+
// Three primary keys, none of which has a docID mapping behind it. Written through the
64+
// same unsafe view the backfill iterates, since the locked wrapper wants a txn in context.
65+
unsafe, ok := stores.Datastore().(interface{ Unsafe() corekv.ReaderWriter })
66+
require.True(t, ok, "the backfill reaches the datastore through Unsafe")
67+
raw := unsafe.Unsafe()
68+
for _, docShortID := range []uint64{1, 2, 3} {
69+
key := keys.PrimaryDataStoreKey{CollectionShortID: shortID, DocShortID: docShortID}
70+
require.NoError(t, raw.Set(ctx, key.Bytes(), []byte{}))
71+
}
72+
73+
p := &P2P{db: multistoreDB{stores: stores}}
74+
col := backfillCollection{collectionID: collectionID}
75+
76+
require.NoError(t, p.pushHeadsForAllDocs(ctx, col, "peer"),
77+
"a primary key with no document must be skipped, not returned as an error")
78+
}

0 commit comments

Comments
 (0)