From 0de48a0cefc27bd77a858084271b0ea87cd952a3 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 11 Aug 2026 18:00:20 +0200 Subject: [PATCH 1/4] keyvalue: prefix deletion watermark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A synced signed row (StoreKeyInner.deletePrefix) whose apply physically drops every row under the prefix older than it — document and ldiff element — and rejects such rows arriving later, so restores from old snapshots cannot resurrect them. The watermark row is the only retained state. Owner-only, on both the local API (Storage.DeletePrefix) and the receive path. Validation now also derives KeyPeerId from the signed payload, closing an impersonation gap: an unchecked id let any writer occupy another peer's row. SYN-145 --- commonspace/object/keyvalue/keyvalue_test.go | 109 +++++++++++++ .../keyvaluestorage/innerstorage/element.go | 29 +++- .../innerstorage/keyvaluestorage.go | 142 ++++++++++++++--- .../innerstorage/keyvaluestorage_test.go | 145 ++++++++++++++++++ .../mock_keyvaluestorage.go | 14 ++ .../keyvalue/keyvaluestorage/storage.go | 90 +++++++++-- .../spacesyncproto/protos/spacesync.proto | 6 + commonspace/spacesyncproto/spacesync.pb.go | 26 +++- .../spacesyncproto/spacesync_drpc.pb.go | 42 ++++- .../spacesyncproto/spacesync_vtproto.pb.go | 43 ++++++ 10 files changed, 602 insertions(+), 44 deletions(-) diff --git a/commonspace/object/keyvalue/keyvalue_test.go b/commonspace/object/keyvalue/keyvalue_test.go index 23ebf0f1c..b92b1198e 100644 --- a/commonspace/object/keyvalue/keyvalue_test.go +++ b/commonspace/object/keyvalue/keyvalue_test.go @@ -450,3 +450,112 @@ func (r *recordingStream) Send(kv *spacesyncproto.StoreKeyValue) error { r.ts.mu.Unlock() return r.DRPCSpaceSync_StoreElementsStream.Send(kv) } + +// rawWatermark hand-builds a signed deletion watermark, letting a test send +// one from an arbitrary identity. +func rawWatermark(t *testing.T, keys *accountdata.AccountKeys, prefix string, ts int64) *spacesyncproto.StoreKeyValue { + protoPeerKey, err := keys.PeerKey.GetPublic().Marshall() + require.NoError(t, err) + protoIdentityKey, err := keys.SignKey.GetPublic().Marshall() + require.NoError(t, err) + inner := spacesyncproto.StoreKeyInner{ + Peer: protoPeerKey, + Identity: protoIdentityKey, + TimestampMicro: ts, + AclHeadId: "acl-head", + Key: prefix, + DeletePrefix: prefix, + } + innerBytes, err := inner.MarshalVT() + require.NoError(t, err) + peerSig, err := keys.PeerKey.Sign(innerBytes) + require.NoError(t, err) + identitySig, err := keys.SignKey.Sign(innerBytes) + require.NoError(t, err) + return &spacesyncproto.StoreKeyValue{ + KeyPeerId: prefix + "-" + keys.PeerKey.GetPublic().PeerId(), + Value: innerBytes, + PeerSignature: peerSig, + IdentitySignature: identitySig, + } +} + +// prefixIds lists the raw stored row ids under a key prefix, watermark rows +// included. +func prefixIds(t *testing.T, store keyvaluestorage.Storage, prefix string) []string { + var ids []string + err := store.InnerStorage().IteratePrefix(ctx, prefix, func(kv innerstorage.KeyValue) error { + ids = append(ids, kv.KeyPeerId) + return nil + }) + require.NoError(t, err) + sort.Strings(ids) + return ids +} + +func TestDeletePrefix(t *testing.T) { + t.Run("drops locally, propagates via sync, rejects resurrection", func(t *testing.T) { + fxClient, fxServer, serverPeer := prepareFixtures(t) + fxClient.add(t, "read/sp1/a", []byte("va")) + fxClient.add(t, "read/sp1/b", []byte("vb")) + fxClient.add(t, "other/c", []byte("vc")) + fxServer.add(t, "read/sp1/d", []byte("vd")) + // syncWithPeer directly: the limiter permits one scheduled sync per + // peer and its Close is terminal, while this test needs two rounds. + require.NoError(t, fxClient.keyValueService.syncWithPeer(ctx, serverPeer)) + require.Len(t, prefixIds(t, fxServer.defaultStore, "read/sp1/"), 3) + + // Capture the pre-delete rows: a device restoring an old snapshot + // would push exactly these. + var stale []*spacesyncproto.StoreKeyValue + err := fxServer.defaultStore.InnerStorage().IteratePrefix(ctx, "read/sp1/", func(kv innerstorage.KeyValue) error { + p := kv.Proto() + p.Value = append([]byte(nil), p.Value...) + p.PeerSignature = append([]byte(nil), p.PeerSignature...) + p.IdentitySignature = append([]byte(nil), p.IdentitySignature...) + stale = append(stale, p) + return nil + }) + require.NoError(t, err) + require.Len(t, stale, 3) + + require.NoError(t, fxClient.defaultStore.DeletePrefix(ctx, "read/sp1/")) + require.NoError(t, fxClient.keyValueService.syncWithPeer(ctx, serverPeer)) + + for _, fx := range []*fixture{fxClient, fxServer} { + ids := prefixIds(t, fx.defaultStore, "read/sp1/") + require.Len(t, ids, 1, "only the watermark row remains: %v", ids) + require.True(t, strings.HasPrefix(ids[0], "read/sp1/-"), "remaining row is the watermark: %v", ids) + require.False(t, fx.check(t, "read/sp1/a", []byte("va"))) + require.False(t, fx.check(t, "read/sp1/d", []byte("vd"))) + require.True(t, fx.check(t, "other/c", []byte("vc")), "rows outside the prefix survive") + // The public read surface hides the watermark row. + var seen []string + require.NoError(t, fx.defaultStore.Iterate(ctx, func(_ keyvaluestorage.Decryptor, key string, _ []innerstorage.KeyValue) (bool, error) { + seen = append(seen, key) + return true, nil + })) + require.Equal(t, []string{"other/c"}, seen) + } + + // Resurrection attempt: replaying the captured pre-delete rows is a + // no-op on both replicas. + require.NoError(t, fxServer.defaultStore.SetRaw(ctx, stale...)) + require.Len(t, prefixIds(t, fxServer.defaultStore, "read/sp1/"), 1) + require.NoError(t, fxClient.defaultStore.SetRaw(ctx, stale...)) + require.Len(t, prefixIds(t, fxClient.defaultStore, "read/sp1/"), 1) + }) + + t.Run("watermark from a non-owner identity is rejected", func(t *testing.T) { + fxClient, _, _ := prepareFixtures(t) + fxClient.add(t, "read/sp1/a", []byte("va")) + + strangerKeys, err := accountdata.NewRandom() + require.NoError(t, err) + wm := rawWatermark(t, strangerKeys, "read/sp1/", time.Now().Add(time.Hour).UnixMicro()) + require.NoError(t, fxClient.defaultStore.SetRaw(ctx, wm)) + + require.True(t, fxClient.check(t, "read/sp1/a", []byte("va")), "rows survive a stranger's watermark") + require.Len(t, prefixIds(t, fxClient.defaultStore, "read/sp1/"), 1, "the stranger's watermark row must not be stored") + }) +} diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go index f550f478f..b5bd72905 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go @@ -9,7 +9,11 @@ import ( "github.com/anyproto/any-sync/util/crypto" ) -var ErrInvalidSignature = errors.New("invalid signature") +var ( + ErrInvalidSignature = errors.New("invalid signature") + ErrInvalidKeyPeerId = errors.New("keyPeerId does not match key and peer") + ErrInvalidWatermark = errors.New("invalid deletion watermark") +) type KeyValue struct { KeyPeerId string @@ -23,6 +27,13 @@ type KeyValue struct { Identity string PeerId string AclId string + // DeletePrefix marks the row as a deletion watermark (see + // StoreKeyInner.deletePrefix). Key mirrors the prefix; the encrypted + // payload stays empty. + DeletePrefix string + // IdentityPubKey is the parsed identity from the inner payload; set by + // KeyValueFromProto for permission checks, never persisted. + IdentityPubKey crypto.PubKey } type Value struct { @@ -50,10 +61,21 @@ func KeyValueFromProto(proto *spacesyncproto.StoreKeyValue, verify bool) (kv Key return kv, err } kv.Identity = identity.Account() + kv.IdentityPubKey = identity kv.PeerId = peerId.PeerId() kv.Key = innerValue.Key kv.AclId = innerValue.AclHeadId - // TODO: check that key-peerId is equal to key+peerId? + kv.DeletePrefix = innerValue.DeletePrefix + // The id must be derived from the signed payload: an unchecked KeyPeerId + // would let any writer occupy (and overwrite) another peer's row. + if kv.KeyPeerId != kv.Key+"-"+kv.PeerId { + return kv, ErrInvalidKeyPeerId + } + // A watermark's key mirrors its prefix, keeping it inside the key range + // it governs (prefix iteration, id-derived invariants). + if kv.DeletePrefix != "" && kv.Key != kv.DeletePrefix { + return kv, ErrInvalidWatermark + } if verify { if verify, _ = identity.Verify(proto.Value, proto.IdentitySignature); !verify { return kv, ErrInvalidSignature @@ -84,6 +106,9 @@ func (kv KeyValue) AnyEnc(a *anyenc.Arena) *anyenc.Value { obj.Set("t", a.NewNumberFloat64(float64(kv.TimestampMicro))) obj.Set("i", a.NewString(kv.Identity)) obj.Set("p", a.NewString(kv.PeerId)) + if kv.DeletePrefix != "" { + obj.Set("dp", a.NewString(kv.DeletePrefix)) + } return obj } diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go index b60d7bfa0..5474419b2 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go @@ -34,6 +34,11 @@ type storage struct { collection anystore.Collection store anystore.DB storageName string + // watermarks maps a deletion prefix to the highest applied watermark + // timestamp: rows under the prefix older than it are dropped and stay + // rejected. Mutated only inside Set, which callers serialize (the outer + // storage holds its mutex around every write). + watermarks map[string]int64 } func New(ctx context.Context, storageName string, headStorage headstorage.HeadStorage, store anystore.DB) (kv KeyValueStorage, err error) { @@ -58,6 +63,7 @@ func New(ctx context.Context, storageName string, headStorage headstorage.HeadSt collection: collection, store: store, diff: ldiff.New(32, 256).(ldiff.CompareDiff), + watermarks: map[string]int64{}, } iter, err := storage.collection.Find(nil).Iter(ctx) if err != nil { @@ -75,6 +81,11 @@ func New(ctx context.Context, storageName string, headStorage headstorage.HeadSt return } elements = append(elements, anyEncToElement(doc.Value())) + if dp := doc.Value().GetString("dp"); dp != "" { + if t := int64(doc.Value().GetFloat64("t")); t > storage.watermarks[dp] { + storage.watermarks[dp] = t + } + } } storage.diff.Set(elements...) hash := storage.diff.Hash() @@ -169,6 +180,7 @@ func (s *storage) keyValueFromDoc(doc anystore.Doc) KeyValue { Identity: doc.Value().GetString("i"), PeerId: doc.Value().GetString("p"), Key: doc.Value().GetString("k"), + DeletePrefix: doc.Value().GetString("dp"), } } @@ -199,9 +211,8 @@ func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { return } var ( - elements, prior []ldiff.Element - added []string - diffUpdated bool + res updateResult + diffUpdated bool ) defer func() { if err == nil { @@ -209,27 +220,42 @@ func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { } else { _ = tx.Rollback() } + if err == nil { + for prefix, t := range res.watermarks { + if t > s.watermarks[prefix] { + s.watermarks[prefix] = t + } + } + } if err != nil && diffUpdated { // The diff already contains this call's elements but the tx did not // commit: undo the in-memory mutations so the diff never advertises // heads the storage doesn't hold — peers would never re-send those // values. Prior heads are restored in reverse so a duplicate id ends - // up at its genuine pre-call state; inserts are then dropped. - for i := len(prior) - 1; i >= 0; i-- { - s.diff.Set(prior[i]) + // up at its genuine pre-call state; inserts are then dropped, and + // watermark-dropped rows re-added (the tx rollback restored their + // documents). + for i := len(res.prior) - 1; i >= 0; i-- { + s.diff.Set(res.prior[i]) } - for _, id := range added { + for _, id := range res.added { _ = s.diff.RemoveId(id) } + for _, el := range res.removed { + s.diff.Set(el) + } } }() ctx = tx.Context() - elements, prior, added, err = s.updateValues(ctx, values...) + res, err = s.updateValues(ctx, values...) if err != nil { return } - s.diff.Set(elements...) - diffUpdated = len(elements) > 0 + s.diff.Set(res.elements...) + for _, el := range res.removed { + _ = s.diff.RemoveId(el.Id) + } + diffUpdated = len(res.elements) > 0 || len(res.removed) > 0 err = s.headStorage.UpdateEntry(ctx, headstorage.HeadsUpdate{ Id: s.storageName, Heads: []string{s.diff.Hash()}, @@ -237,18 +263,52 @@ func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { return } -// updateValues upserts the values that win their LWW comparison and returns the -// new diff elements along with what is needed to undo the diff on a failed tx: -// the replaced elements' prior state and the ids that were not present before. -func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (elements, prior []ldiff.Element, added []string, err error) { +// updateResult carries one updateValues batch outcome: the new diff elements, +// what is needed to undo the diff on a failed tx (replaced elements' prior +// state, inserted ids, elements of watermark-dropped rows), and the watermarks +// to publish into s.watermarks once the tx commits. +type updateResult struct { + elements []ldiff.Element + prior []ldiff.Element + added []string + removed []ldiff.Element + watermarks map[string]int64 +} + +// covered reports whether a row loses to an applied or in-batch watermark: +// its key falls under the prefix and it is strictly older. A watermark row +// itself never loses to its own prefix (equal timestamps are not covered). +func (s *storage) covered(res *updateResult, key string, timestampMicro int64) bool { + for prefix, t := range s.watermarks { + if timestampMicro < t && strings.HasPrefix(key, prefix) { + return true + } + } + for prefix, t := range res.watermarks { + if timestampMicro < t && strings.HasPrefix(key, prefix) { + return true + } + } + return false +} + +// updateValues upserts the values that win their LWW comparison. A winning +// watermark value additionally drops every stored row under its prefix that +// is older than it — document and diff element — leaving the watermark row +// as the only retained state. +func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (res updateResult, err error) { parser := parserPool.Get() defer parserPool.Put(parser) arena := arenaPool.Get() defer arenaPool.Put(arena) - elements = make([]ldiff.Element, 0, len(values)) + res.elements = make([]ldiff.Element, 0, len(values)) + res.watermarks = map[string]int64{} var doc anystore.Doc for _, value := range values { + if s.covered(&res, value.Key, value.TimestampMicro) { + continue + } doc, err = s.collection.FindIdWithParser(ctx, parser, value.KeyPeerId) isNotFound := errors.Is(err, anystore.ErrDocNotFound) if err != nil && !isNotFound { @@ -258,20 +318,66 @@ func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (element if int64(doc.Value().GetFloat64("t")) >= value.TimestampMicro { continue } - prior = append(prior, anyEncToElement(doc.Value())) + res.prior = append(res.prior, anyEncToElement(doc.Value())) } else { - added = append(added, value.KeyPeerId) + res.added = append(res.added, value.KeyPeerId) } arena.Reset() val := value.AnyEnc(arena) if err = s.collection.UpsertOne(ctx, val); err != nil { return } - elements = append(elements, anyEncToElement(val)) + res.elements = append(res.elements, anyEncToElement(val)) + if value.DeletePrefix != "" { + if err = s.applyWatermark(ctx, &res, value); err != nil { + return + } + } } return } +// applyWatermark physically deletes every stored row whose key starts with +// the watermark's prefix and is strictly older than it, the just-written +// watermark row excepted by the timestamp comparison. Runs inside Set's tx; +// diff elements of the dropped rows are removed by Set after it returns. +func (s *storage) applyWatermark(ctx context.Context, res *updateResult, wm KeyValue) (err error) { + // KeyPeerId starts with Key, so a key-prefix scan is an id-prefix scan. + filter := query.Key{Path: []string{"id"}, Filter: query.NewComp(query.CompOpGte, wm.DeletePrefix)} + iter, err := s.collection.Find(filter).Sort("id").Iter(ctx) + if err != nil { + return err + } + var dropIds []string + var doc anystore.Doc + for iter.Next() { + if doc, err = iter.Doc(); err != nil { + _ = iter.Close() + return err + } + if !strings.HasPrefix(doc.Value().GetString("id"), wm.DeletePrefix) { + break + } + if int64(doc.Value().GetFloat64("t")) >= wm.TimestampMicro { + continue + } + res.removed = append(res.removed, anyEncToElement(doc.Value())) + dropIds = append(dropIds, doc.Value().GetString("id")) + } + if err = iter.Close(); err != nil { + return err + } + for _, id := range dropIds { + if err = s.collection.DeleteId(ctx, id); err != nil { + return err + } + } + if wm.TimestampMicro > res.watermarks[wm.DeletePrefix] { + res.watermarks[wm.DeletePrefix] = wm.TimestampMicro + } + return nil +} + func anyEncToElement(val *anyenc.Value) ldiff.Element { byteRepr := make([]byte, 8) binary.BigEndian.PutUint64(byteRepr, uint64(int64(val.GetFloat64("t")))) diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go index 5cd5fff14..c3337637c 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go @@ -162,3 +162,148 @@ func TestIterateValuesReturnsOwnedMemory(t *testing.T) { require.Equal(t, want.Value.IdentitySignature, kv.Value.IdentitySignature, "identity signature of %s must survive iteration", kv.KeyPeerId) } } + +// rowKV builds a normal row for watermark tests: id derived from key+peer, +// explicit timestamp. +func rowKV(key, peer string, ts int64) innerstorage.KeyValue { + return innerstorage.KeyValue{ + KeyPeerId: key + "-" + peer, + Key: key, + ReadKeyId: "readKeyId", + Identity: "identity", + PeerId: peer, + TimestampMicro: ts, + Value: innerstorage.Value{ + Value: []byte("v-" + key + "-" + peer), + PeerSignature: []byte("ps"), + IdentitySignature: []byte("is"), + }, + } +} + +// wmKV builds a deletion watermark row for the prefix. +func wmKV(prefix, peer string, ts int64) innerstorage.KeyValue { + kv := rowKV(prefix, peer, ts) + kv.DeletePrefix = prefix + kv.Value.Value = nil + return kv +} + +func storedIds(t *testing.T, storage innerstorage.KeyValueStorage) []string { + var ids []string + require.NoError(t, storage.IterateValues(ctx, func(kv innerstorage.KeyValue) (bool, error) { + ids = append(ids, kv.KeyPeerId) + return true, nil + })) + sort.Strings(ids) + return ids +} + +// TestWatermarkDropsOlderRows: applying a watermark physically removes every +// row under the prefix older than it — document and diff element — while +// newer rows and rows outside the prefix survive. +func TestWatermarkDropsOlderRows(t *testing.T) { + storage := newTestStorage(t) + require.NoError(t, storage.Set(ctx, + rowKV("read/sp1/obj1", "peerA", 10), + rowKV("read/sp1/obj1", "peerB", 11), + rowKV("read/sp1/obj2", "peerA", 12), + rowKV("read/sp1/obj3", "peerA", 200), // newer than the watermark + rowKV("read/sp2/obj1", "peerA", 13), // outside the prefix + )) + hashBefore := storage.Diff().Hash() + + wm := wmKV("read/sp1/", "peerA", 100) + require.NoError(t, storage.Set(ctx, wm)) + + require.Equal(t, []string{ + "read/sp1/-peerA", // the watermark row itself + "read/sp1/obj3-peerA", + "read/sp2/obj1-peerA", + }, storedIds(t, storage)) + for _, dropped := range []string{"read/sp1/obj1-peerA", "read/sp1/obj1-peerB", "read/sp1/obj2-peerA"} { + _, err := storage.Diff().Element(dropped) + require.ErrorIs(t, err, ldiff.ErrElementNotFound, dropped) + } + _, err := storage.Diff().Element(wm.KeyPeerId) + require.NoError(t, err, "watermark element must be advertised") + require.NotEqual(t, hashBefore, storage.Diff().Hash()) +} + +// TestWatermarkRejectsLateArrivals: rows under the prefix older than an +// applied watermark are discarded on arrival; equal-or-newer rows apply. +func TestWatermarkRejectsLateArrivals(t *testing.T) { + storage := newTestStorage(t) + require.NoError(t, storage.Set(ctx, wmKV("read/sp1/", "peerA", 100))) + + require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj1", "peerB", 99))) + _, err := storage.GetKeyPeerId(ctx, "read/sp1/obj1-peerB") + require.ErrorIs(t, err, anystore.ErrDocNotFound, "older row must be rejected") + + require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj2", "peerB", 100))) + _, err = storage.GetKeyPeerId(ctx, "read/sp1/obj2-peerB") + require.NoError(t, err, "equal-timestamp row is not covered") +} + +// TestWatermarkSurvivesReopen: the watermark index is rebuilt from the stored +// watermark row, so rejection keeps working after a restart. +func TestWatermarkSurvivesReopen(t *testing.T) { + db, err := anystore.Open(ctx, filepath.Join(t.TempDir(), "store.db"), nil) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + heads, err := headstorage.New(ctx, db) + require.NoError(t, err) + storage, err := innerstorage.New(ctx, "kv.test", heads, db) + require.NoError(t, err) + require.NoError(t, storage.Set(ctx, wmKV("read/sp1/", "peerA", 100))) + + reopened, err := innerstorage.New(ctx, "kv.test", heads, db) + require.NoError(t, err) + require.NoError(t, reopened.Set(ctx, rowKV("read/sp1/obj1", "peerB", 99))) + _, err = reopened.GetKeyPeerId(ctx, "read/sp1/obj1-peerB") + require.ErrorIs(t, err, anystore.ErrDocNotFound, "watermark must reject after reopen") +} + +// TestWatermarkBatchOrderIndependent: a batch carrying both an old row and a +// watermark covering it converges to the same state in either order. +func TestWatermarkBatchOrderIndependent(t *testing.T) { + for name, batch := range map[string][]innerstorage.KeyValue{ + "row first": {rowKV("read/sp1/obj1", "peerB", 50), wmKV("read/sp1/", "peerA", 100)}, + "wm first": {wmKV("read/sp1/", "peerA", 100), rowKV("read/sp1/obj1", "peerB", 50)}, + } { + t.Run(name, func(t *testing.T) { + storage := newTestStorage(t) + require.NoError(t, storage.Set(ctx, batch...)) + require.Equal(t, []string{"read/sp1/-peerA"}, storedIds(t, storage)) + }) + } +} + +// TestWatermarkFailedTxRestoresDroppedElements: when the tx fails after a +// watermark dropped rows, the diff must re-advertise them — the documents +// come back with the rollback. +func TestWatermarkFailedTxRestoresDroppedElements(t *testing.T) { + storage, failingHeads := newTestStorageWithFailingHeads(t) + row := rowKV("read/sp1/obj1", "peerB", 50) + require.NoError(t, storage.Set(ctx, row)) + hashBefore := storage.Diff().Hash() + + failingHeads.fail = true + wm := wmKV("read/sp1/", "peerA", 100) + require.Error(t, storage.Set(ctx, wm)) + failingHeads.fail = false + + _, err := storage.Diff().Element(row.KeyPeerId) + require.NoError(t, err, "dropped element must be restored on rollback") + _, err = storage.Diff().Element(wm.KeyPeerId) + require.ErrorIs(t, err, ldiff.ErrElementNotFound, "rolled-back watermark must not be advertised") + require.Equal(t, hashBefore, storage.Diff().Hash()) + _, err = storage.GetKeyPeerId(ctx, row.KeyPeerId) + require.NoError(t, err, "row document must survive the rollback") + + // The rolled-back watermark must not keep rejecting: a fresh old row + // still applies. + require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj2", "peerB", 60))) + _, err = storage.GetKeyPeerId(ctx, "read/sp1/obj2-peerB") + require.NoError(t, err) +} diff --git a/commonspace/object/keyvalue/keyvaluestorage/mock_keyvaluestorage/mock_keyvaluestorage.go b/commonspace/object/keyvalue/keyvaluestorage/mock_keyvaluestorage/mock_keyvaluestorage.go index 19af64f03..8d91de6ea 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/mock_keyvaluestorage/mock_keyvaluestorage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/mock_keyvaluestorage/mock_keyvaluestorage.go @@ -43,6 +43,20 @@ func (m *MockStorage) EXPECT() *MockStorageMockRecorder { return m.recorder } +// DeletePrefix mocks base method. +func (m *MockStorage) DeletePrefix(ctx context.Context, prefix string) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeletePrefix", ctx, prefix) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeletePrefix indicates an expected call of DeletePrefix. +func (mr *MockStorageMockRecorder) DeletePrefix(ctx, prefix any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeletePrefix", reflect.TypeOf((*MockStorage)(nil).DeletePrefix), ctx, prefix) +} + // GetAll mocks base method. func (m *MockStorage) GetAll(ctx context.Context, key string, get func(keyvaluestorage.Decryptor, []innerstorage.KeyValue) error) error { m.ctrl.T.Helper() diff --git a/commonspace/object/keyvalue/keyvaluestorage/storage.go b/commonspace/object/keyvalue/keyvaluestorage/storage.go index 3ae797d69..1a58d195e 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/storage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/storage.go @@ -53,6 +53,12 @@ type Storage interface { Prepare() error Set(ctx context.Context, key string, value []byte) error SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreKeyValue) error + // DeletePrefix publishes a deletion watermark: every row whose key + // starts with prefix and predates the watermark is physically dropped + // on every replica, and stays rejected should it arrive again (e.g. + // from a device restoring an old snapshot). The watermark row is the + // only retained state. Owner-only. + DeletePrefix(ctx context.Context, prefix string) error GetAll(ctx context.Context, key string, get func(decryptor Decryptor, values []innerstorage.KeyValue) error) error Iterate(ctx context.Context, f func(decryptor Decryptor, key string, values []innerstorage.KeyValue) (bool, error)) error InnerStorage() innerstorage.KeyValueStorage @@ -111,24 +117,51 @@ func (s *storage) Id() string { func (s *storage) Set(ctx context.Context, key string, value []byte) error { s.mx.Lock() defer s.mx.Unlock() - s.aclList.RLock() - headId := s.aclList.Head().Id - state := s.aclList.AclState() - if !s.aclList.AclState().Permissions(state.Identity()).CanWrite() { - s.aclList.RUnlock() - return list.ErrInsufficientPermissions - } - readKeyId := state.CurrentReadKeyId() - err := s.readKeysFromAclState(state) + headId, readKeyId, err := s.prepareWrite(func(p list.AclPermissions) bool { return p.CanWrite() }) if err != nil { - s.aclList.RUnlock() return err } - s.aclList.RUnlock() value, err = s.currentReadKey.Encrypt(value) if err != nil { return err } + return s.signAndStore(ctx, key, value, "", headId, readKeyId) +} + +func (s *storage) DeletePrefix(ctx context.Context, prefix string) error { + s.mx.Lock() + defer s.mx.Unlock() + // Owner-only: a watermark removes rows written by every peer, not just + // the caller's own. + headId, readKeyId, err := s.prepareWrite(func(p list.AclPermissions) bool { return p.IsOwner() }) + if err != nil { + return err + } + return s.signAndStore(ctx, prefix, nil, prefix, headId, readKeyId) +} + +// prepareWrite runs the shared ACL section of a local write: permission +// check, read-key refresh, and the current head/read-key ids the row is +// stamped with. +func (s *storage) prepareWrite(allowed func(list.AclPermissions) bool) (headId, readKeyId string, err error) { + s.aclList.RLock() + defer s.aclList.RUnlock() + headId = s.aclList.Head().Id + state := s.aclList.AclState() + if !allowed(state.Permissions(state.Identity())) { + return "", "", list.ErrInsufficientPermissions + } + readKeyId = state.CurrentReadKeyId() + if err = s.readKeysFromAclState(state); err != nil { + return "", "", err + } + return headId, readKeyId, nil +} + +// signAndStore builds, signs, applies and broadcasts one own row: a regular +// value (encrypted by the caller) or, with deletePrefix set, a deletion +// watermark (empty value, key mirrors the prefix). +func (s *storage) signAndStore(ctx context.Context, key string, value []byte, deletePrefix string, headId, readKeyId string) error { peerIdKey := s.keys.PeerKey identityKey := s.keys.SignKey protoPeerKey, err := peerIdKey.GetPublic().Marshall() @@ -147,6 +180,7 @@ func (s *storage) Set(ctx context.Context, key string, value []byte) error { TimestampMicro: timestampMicro, AclHeadId: headId, Key: key, + DeletePrefix: deletePrefix, } innerBytes, err := inner.MarshalVT() if err != nil { @@ -169,6 +203,7 @@ func (s *storage) Set(ctx context.Context, key string, value []byte) error { PeerId: peerIdKey.GetPublic().PeerId(), AclId: headId, ReadKeyId: readKeyId, + DeletePrefix: deletePrefix, Value: innerstorage.Value{ Value: innerBytes, PeerSignature: peerSig, @@ -179,9 +214,11 @@ func (s *storage) Set(ctx context.Context, key string, value []byte) error { if err != nil { return err } - indexErr := s.indexer.Index(s.decrypt, keyValue) - if indexErr != nil { - log.Warn("failed to index for key", zap.String("key", key), zap.Error(indexErr)) + if deletePrefix == "" { + indexErr := s.indexer.Index(s.decrypt, keyValue) + if indexErr != nil { + log.Warn("failed to index for key", zap.String("key", key), zap.Error(indexErr)) + } } sendErr := s.syncClient.Broadcast(ctx, s.storageId, keyValue) if sendErr != nil { @@ -217,6 +254,12 @@ func (s *storage) SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreK return err } for i := range keyValues { + // A watermark deletes other peers' rows, so accepting one demands + // the owner's identity — the same bar DeletePrefix applies locally. + if keyValues[i].DeletePrefix != "" && !state.Permissions(keyValues[i].IdentityPubKey).IsOwner() { + keyValues[i].KeyPeerId = "" + continue + } el, err := s.inner.Diff().Element(keyValues[i].KeyPeerId) if err == nil { binary.BigEndian.PutUint64(s.byteRepr, uint64(keyValues[i].TimestampMicro)) @@ -246,9 +289,16 @@ func (s *storage) SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreK if sendErr != nil { log.Warn("failed to send key values", zap.Error(sendErr)) } - indexErr := s.indexer.Index(s.decrypt, keyValues...) - if indexErr != nil { - log.Warn("failed to index for keys", zap.Error(indexErr)) + // Watermarks carry no payload to index; their effect (dropped rows) is + // already applied. + indexable := slice.DiscardFromSlice(keyValues, func(value innerstorage.KeyValue) bool { + return value.DeletePrefix != "" + }) + if len(indexable) > 0 { + indexErr := s.indexer.Index(s.decrypt, indexable...) + if indexErr != nil { + log.Warn("failed to index for keys", zap.Error(indexErr)) + } } return nil } @@ -256,6 +306,9 @@ func (s *storage) SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreK func (s *storage) GetAll(ctx context.Context, key string, get func(decryptor Decryptor, values []innerstorage.KeyValue) error) (err error) { var values []innerstorage.KeyValue err = s.inner.IteratePrefix(ctx, key, func(kv innerstorage.KeyValue) error { + if kv.DeletePrefix != "" { + return nil + } values = append(values, kv) return nil }) @@ -317,6 +370,9 @@ func (s *storage) Iterate(ctx context.Context, f func(decryptor Decryptor, key s values []innerstorage.KeyValue ) err = s.inner.IterateValues(ctx, func(kv innerstorage.KeyValue) (bool, error) { + if kv.DeletePrefix != "" { + return true, nil + } if kv.Key != curKey { if curKey != "" { iter, err := f(s.decrypt, curKey, values) diff --git a/commonspace/spacesyncproto/protos/spacesync.proto b/commonspace/spacesyncproto/protos/spacesync.proto index 21c9b7bd2..af20330c7 100644 --- a/commonspace/spacesyncproto/protos/spacesync.proto +++ b/commonspace/spacesyncproto/protos/spacesync.proto @@ -262,6 +262,12 @@ message StoreKeyInner { int64 timestampMicro = 4; string aclHeadId = 5; string key = 6; + // deletePrefix marks the row as a deletion watermark: applying it + // physically removes every stored row whose key starts with the + // prefix and whose timestamp is older than timestampMicro, and + // rejects such rows arriving later. The watermark row itself is the + // only retained state. key mirrors the prefix; value stays empty. + string deletePrefix = 7; } message StorageHeader { diff --git a/commonspace/spacesyncproto/spacesync.pb.go b/commonspace/spacesyncproto/spacesync.pb.go index f18393ebd..2ecde95ec 100644 --- a/commonspace/spacesyncproto/spacesync.pb.go +++ b/commonspace/spacesyncproto/spacesync.pb.go @@ -240,8 +240,8 @@ type DiffType int32 const ( DiffType_Initial DiffType = 0 - DiffType_V1 DiffType = 1 - DiffType_V2 DiffType = 2 + DiffType_V1 DiffType = 1 // deprecated, not supported anymore + DiffType_V2 DiffType = 2 // deprecated, not supported anymore DiffType_V3 DiffType = 3 ) @@ -2075,8 +2075,14 @@ type StoreKeyInner struct { TimestampMicro int64 `protobuf:"varint,4,opt,name=timestampMicro,proto3" json:"timestampMicro,omitempty"` AclHeadId string `protobuf:"bytes,5,opt,name=aclHeadId,proto3" json:"aclHeadId,omitempty"` Key string `protobuf:"bytes,6,opt,name=key,proto3" json:"key,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // deletePrefix marks the row as a deletion watermark: applying it + // physically removes every stored row whose key starts with the + // prefix and whose timestamp is older than timestampMicro, and + // rejects such rows arriving later. The watermark row itself is the + // only retained state. key mirrors the prefix; value stays empty. + DeletePrefix string `protobuf:"bytes,7,opt,name=deletePrefix,proto3" json:"deletePrefix,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *StoreKeyInner) Reset() { @@ -2151,6 +2157,13 @@ func (x *StoreKeyInner) GetKey() string { return "" } +func (x *StoreKeyInner) GetDeletePrefix() string { + if x != nil { + return x.DeletePrefix + } + return "" +} + type StorageHeader struct { state protoimpl.MessageState `protogen:"open.v1"` SpaceId string `protobuf:"bytes,1,opt,name=spaceId,proto3" json:"spaceId,omitempty"` @@ -2326,14 +2339,15 @@ const file_commonspace_spacesyncproto_protos_spacesync_proto_rawDesc = "" + "\rpeerSignature\x18\x04 \x01(\fR\rpeerSignature\x12\x18\n" + "\aspaceId\x18\x05 \x01(\tR\aspaceId\"H\n" + "\x0eStoreKeyValues\x126\n" + - "\tkeyValues\x18\x01 \x03(\v2\x18.spacesync.StoreKeyValueR\tkeyValues\"\xad\x01\n" + + "\tkeyValues\x18\x01 \x03(\v2\x18.spacesync.StoreKeyValueR\tkeyValues\"\xd1\x01\n" + "\rStoreKeyInner\x12\x12\n" + "\x04peer\x18\x01 \x01(\fR\x04peer\x12\x1a\n" + "\bidentity\x18\x02 \x01(\fR\bidentity\x12\x14\n" + "\x05value\x18\x03 \x01(\fR\x05value\x12&\n" + "\x0etimestampMicro\x18\x04 \x01(\x03R\x0etimestampMicro\x12\x1c\n" + "\taclHeadId\x18\x05 \x01(\tR\taclHeadId\x12\x10\n" + - "\x03key\x18\x06 \x01(\tR\x03key\"K\n" + + "\x03key\x18\x06 \x01(\tR\x03key\x12\"\n" + + "\fdeletePrefix\x18\a \x01(\tR\fdeletePrefix\"K\n" + "\rStorageHeader\x12\x18\n" + "\aspaceId\x18\x01 \x01(\tR\aspaceId\x12 \n" + "\vstorageName\x18\x02 \x01(\tR\vstorageName*\xee\x01\n" + diff --git a/commonspace/spacesyncproto/spacesync_drpc.pb.go b/commonspace/spacesyncproto/spacesync_drpc.pb.go index 944aec717..d1b346735 100644 --- a/commonspace/spacesyncproto/spacesync_drpc.pb.go +++ b/commonspace/spacesyncproto/spacesync_drpc.pb.go @@ -1,5 +1,5 @@ // Code generated by protoc-gen-go-drpc. DO NOT EDIT. -// protoc-gen-go-drpc version: v0.0.34 +// protoc-gen-go-drpc version: v1.0.0 // source: commonspace/spacesyncproto/protos/spacesync.proto package spacesyncproto @@ -403,6 +403,10 @@ type drpcSpaceSync_HeadSyncStream struct { drpc.Stream } +func (x *drpcSpaceSync_HeadSyncStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_HeadSyncStream) SendAndClose(m *HeadSyncResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -419,6 +423,10 @@ type drpcSpaceSync_StoreDiffStream struct { drpc.Stream } +func (x *drpcSpaceSync_StoreDiffStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_StoreDiffStream) SendAndClose(m *StoreDiffResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -436,6 +444,10 @@ type drpcSpaceSync_StoreElementsStream struct { drpc.Stream } +func (x *drpcSpaceSync_StoreElementsStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_StoreElementsStream) Send(m *StoreKeyValue) error { return x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}) } @@ -461,6 +473,10 @@ type drpcSpaceSync_SpacePushStream struct { drpc.Stream } +func (x *drpcSpaceSync_SpacePushStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_SpacePushStream) SendAndClose(m *SpacePushResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -477,6 +493,10 @@ type drpcSpaceSync_SpacePullStream struct { drpc.Stream } +func (x *drpcSpaceSync_SpacePullStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_SpacePullStream) SendAndClose(m *SpacePullResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -494,6 +514,10 @@ type drpcSpaceSync_ObjectSyncStreamStream struct { drpc.Stream } +func (x *drpcSpaceSync_ObjectSyncStreamStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_ObjectSyncStreamStream) Send(m *ObjectSyncMessage) error { return x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}) } @@ -519,6 +543,10 @@ type drpcSpaceSync_ObjectSyncStream struct { drpc.Stream } +func (x *drpcSpaceSync_ObjectSyncStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_ObjectSyncStream) SendAndClose(m *ObjectSyncMessage) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -535,6 +563,10 @@ type drpcSpaceSync_ObjectSyncRequestStreamStream struct { drpc.Stream } +func (x *drpcSpaceSync_ObjectSyncRequestStreamStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_ObjectSyncRequestStreamStream) Send(m *ObjectSyncMessage) error { return x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}) } @@ -548,6 +580,10 @@ type drpcSpaceSync_AclAddRecordStream struct { drpc.Stream } +func (x *drpcSpaceSync_AclAddRecordStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_AclAddRecordStream) SendAndClose(m *AclAddRecordResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err @@ -564,6 +600,10 @@ type drpcSpaceSync_AclGetRecordsStream struct { drpc.Stream } +func (x *drpcSpaceSync_AclGetRecordsStream) GetStream() drpc.Stream { + return x.Stream +} + func (x *drpcSpaceSync_AclGetRecordsStream) SendAndClose(m *AclGetRecordsResponse) error { if err := x.MsgSend(m, drpcEncoding_File_commonspace_spacesyncproto_protos_spacesync_proto{}); err != nil { return err diff --git a/commonspace/spacesyncproto/spacesync_vtproto.pb.go b/commonspace/spacesyncproto/spacesync_vtproto.pb.go index 1f1ae6093..c089704fb 100644 --- a/commonspace/spacesyncproto/spacesync_vtproto.pb.go +++ b/commonspace/spacesyncproto/spacesync_vtproto.pb.go @@ -1628,6 +1628,13 @@ func (m *StoreKeyInner) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.DeletePrefix) > 0 { + i -= len(m.DeletePrefix) + copy(dAtA[i:], m.DeletePrefix) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.DeletePrefix))) + i-- + dAtA[i] = 0x3a + } if len(m.Key) > 0 { i -= len(m.Key) copy(dAtA[i:], m.Key) @@ -2371,6 +2378,10 @@ func (m *StoreKeyInner) SizeVT() (n int) { if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + l = len(m.DeletePrefix) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -6425,6 +6436,38 @@ func (m *StoreKeyInner) UnmarshalVT(dAtA []byte) error { } m.Key = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 7: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DeletePrefix", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DeletePrefix = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) From 42e3b31175b601da305202afca279e4effdec4c0 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 11 Aug 2026 18:39:28 +0200 Subject: [PATCH 2/4] keyvalue: review fixes for the deletion watermark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - applyWatermark filters on the row's key, not its id: an id can match a prefix past its key's end (key+"-"+peerId), which deleted rows of a shorter key and diverged replicas from the covered() rejection rule - failed-tx diff undo restores a per-id pre-call snapshot instead of replaying prior/added/removed lists, which advertised phantom heads when a row was written and watermark-dropped in the same batch - inner Set returns SetResult (applied values + dropped ids): SetRaw now broadcasts and indexes only applied rows — indexing store-rejected rows resurrected deleted data downstream - optional DeletionAwareIndexer.RemoveIndex mirrors watermark drops into the app index - local Set/DeletePrefix under a newer covering watermark fail with ErrCoveredByWatermark instead of silently succeeding while every replica discards the row - DeletePrefix("") rejected - syncWithPeer tolerates a row deleted between CompareDiff and its load - IteratePrefix and applyWatermark share one id-prefix iterator; the break condition is HasPrefix (Contains could over-include mid-id matches past the range) SYN-145 --- commonspace/object/keyvalue/keyvalue.go | 6 + commonspace/object/keyvalue/keyvalue_test.go | 127 +++++++++++++ .../innerstorage/keyvaluestorage.go | 173 +++++++++++------- .../innerstorage/keyvaluestorage_test.go | 101 ++++++++-- .../keyvalue/keyvaluestorage/storage.go | 59 +++++- 5 files changed, 378 insertions(+), 88 deletions(-) diff --git a/commonspace/object/keyvalue/keyvalue.go b/commonspace/object/keyvalue/keyvalue.go index 01ca32594..7f9000539 100644 --- a/commonspace/object/keyvalue/keyvalue.go +++ b/commonspace/object/keyvalue/keyvalue.go @@ -6,6 +6,7 @@ import ( "slices" "strings" + anystore "github.com/anyproto/any-store" "go.uber.org/zap" "storj.io/drpc" @@ -98,6 +99,11 @@ func (k *keyValueService) syncWithPeer(ctx context.Context, p peer.Peer) (err er for _, id := range append(removedIds, changedIds...) { kv, err := innerStorage.GetKeyPeerId(ctx, id) if err != nil { + if errors.Is(err, anystore.ErrDocNotFound) { + // A concurrently applied deletion watermark dropped the row + // between CompareDiff and this load; skip it for the round. + continue + } return err } err = stream.Send(kv.Proto()) diff --git a/commonspace/object/keyvalue/keyvalue_test.go b/commonspace/object/keyvalue/keyvalue_test.go index b92b1198e..6469c9e8b 100644 --- a/commonspace/object/keyvalue/keyvalue_test.go +++ b/commonspace/object/keyvalue/keyvalue_test.go @@ -18,6 +18,7 @@ import ( anystore "github.com/anyproto/any-store" "github.com/stretchr/testify/require" + "github.com/anyproto/any-sync/app" "github.com/anyproto/any-sync/commonspace/object/accountdata" "github.com/anyproto/any-sync/commonspace/object/acl/list" "github.com/anyproto/any-sync/commonspace/object/acl/recordverifier" @@ -559,3 +560,129 @@ func TestDeletePrefix(t *testing.T) { require.Len(t, prefixIds(t, fxClient.defaultStore, "read/sp1/"), 1, "the stranger's watermark row must not be stored") }) } + +// recordingIndexer records Index/RemoveIndex calls; RemoveIndex makes it +// deletion-aware. +type recordingIndexer struct { + mu sync.Mutex + indexed []string + removed []string +} + +func (r *recordingIndexer) Init(a *app.App) error { return nil } +func (r *recordingIndexer) Name() string { return keyvaluestorage.IndexerCName } + +func (r *recordingIndexer) Index(_ keyvaluestorage.Decryptor, keyValues ...innerstorage.KeyValue) error { + r.mu.Lock() + defer r.mu.Unlock() + for _, kv := range keyValues { + r.indexed = append(r.indexed, kv.KeyPeerId) + } + return nil +} + +func (r *recordingIndexer) RemoveIndex(keyPeerIds ...string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.removed = append(r.removed, keyPeerIds...) + return nil +} + +func (r *recordingIndexer) counts() (indexed, removed int) { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.indexed), len(r.removed) +} + +// newBareStorage builds a keyvaluestorage.Storage without the rpc service +// around it, with an arbitrary indexer. +func newBareStorage(t *testing.T, keys *accountdata.AccountKeys, spacePayload spacestorage.SpaceStorageCreatePayload, indexer keyvaluestorage.Indexer) keyvaluestorage.Storage { + anyStore, err := anystore.Open(ctx, filepath.Join(t.TempDir(), "store.db"), nil) + require.NoError(t, err) + t.Cleanup(func() { _ = anyStore.Close() }) + storage, err := spacestorage.Create(ctx, anyStore, spacePayload) + require.NoError(t, err) + aclStorage, err := storage.AclStorage() + require.NoError(t, err) + aclList, err := list.BuildAclListWithIdentity(keys, aclStorage, recordverifier.NewValidateFull()) + require.NoError(t, err) + st, err := keyvaluestorage.New(ctx, "kv.storage", anyStore, storage.HeadStorage(), keys, &countingSyncClient{}, aclList, indexer) + require.NoError(t, err) + return st +} + +// exportPrefix returns the wire protos of every raw row under the prefix. +func exportPrefix(t *testing.T, store keyvaluestorage.Storage, prefix string) []*spacesyncproto.StoreKeyValue { + var protos []*spacesyncproto.StoreKeyValue + err := store.InnerStorage().IteratePrefix(ctx, prefix, func(kv innerstorage.KeyValue) error { + protos = append(protos, kv.Proto()) + return nil + }) + require.NoError(t, err) + return protos +} + +// TestSetCoveredByWatermark: a local write under a newer watermark fails +// loudly instead of returning success for a value every replica would drop. +func TestSetCoveredByWatermark(t *testing.T) { + fxClient, _, _ := prepareFixtures(t) + _, err := fxClient.defaultStore.InnerStorage().Set(ctx, innerstorage.KeyValue{ + KeyPeerId: "read/sp1/-peerX", + Key: "read/sp1/", + PeerId: "peerX", + Identity: "identity", + DeletePrefix: "read/sp1/", + TimestampMicro: time.Now().Add(time.Hour).UnixMicro(), + }) + require.NoError(t, err) + + err = fxClient.defaultStore.Set(ctx, "read/sp1/a", []byte("va")) + require.ErrorIs(t, err, keyvaluestorage.ErrCoveredByWatermark) + require.ErrorIs(t, fxClient.defaultStore.DeletePrefix(ctx, "read/sp1/sub/"), keyvaluestorage.ErrCoveredByWatermark, + "a narrower watermark under a newer covering one is refused too") + require.NoError(t, fxClient.defaultStore.Set(ctx, "other/b", []byte("vb")), "keys outside the prefix write normally") +} + +func TestDeletePrefixEmpty(t *testing.T) { + fxClient, _, _ := prepareFixtures(t) + require.ErrorIs(t, fxClient.defaultStore.DeletePrefix(ctx, ""), keyvaluestorage.ErrEmptyDeletePrefix) +} + +// TestIndexerDeletionFlow pins the indexer contract around watermarks: +// applied rows index, watermark-dropped rows un-index via the optional +// DeletionAwareIndexer, and rows the store rejects never reach Index. +func TestIndexerDeletionFlow(t *testing.T) { + firstKeys, err := accountdata.NewRandom() + require.NoError(t, err) + secondKeys, err := accountdata.NewRandom() + require.NoError(t, err) + secondKeys.SignKey = firstKeys.SignKey + payload := newStorageCreatePayload(t, firstKeys) + + idx := &recordingIndexer{} + stA := newBareStorage(t, firstKeys, payload, keyvaluestorage.NoOpIndexer{}) + stB := newBareStorage(t, secondKeys, payload, idx) + + require.NoError(t, stA.Set(ctx, "read/sp1/a", []byte("va"))) + require.NoError(t, stA.Set(ctx, "read/sp1/b", []byte("vb"))) + stale := exportPrefix(t, stA, "read/sp1/") + require.Len(t, stale, 2) + + require.NoError(t, stB.SetRaw(ctx, stale...)) + indexed, removed := idx.counts() + require.Equal(t, 2, indexed, "applied rows index") + require.Equal(t, 0, removed) + + require.NoError(t, stA.DeletePrefix(ctx, "read/sp1/")) + wmProtos := exportPrefix(t, stA, "read/sp1/") + require.Len(t, wmProtos, 1, "only the watermark row remains on A") + require.NoError(t, stB.SetRaw(ctx, wmProtos...)) + indexed, removed = idx.counts() + require.Equal(t, 2, indexed, "the watermark row itself is not indexed") + require.Equal(t, 2, removed, "dropped rows are un-indexed") + + // Resurrection attempt: the rejected rows must not reach Index. + require.NoError(t, stB.SetRaw(ctx, stale...)) + indexed, _ = idx.counts() + require.Equal(t, 2, indexed, "rejected rows must not be indexed") +} diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go index 5474419b2..feee61cbb 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage.go @@ -21,11 +21,24 @@ var ( ) type KeyValueStorage interface { - Set(ctx context.Context, keyValues ...KeyValue) (err error) + Set(ctx context.Context, keyValues ...KeyValue) (result SetResult, err error) Diff() ldiff.CompareDiff GetKeyPeerId(ctx context.Context, keyPeerId string) (keyValue KeyValue, err error) IterateValues(context.Context, func(kv KeyValue) (bool, error)) (err error) IteratePrefix(context.Context, string, func(kv KeyValue) error) (err error) + // WatermarkTs returns the highest applied watermark timestamp covering + // key, 0 when none. Serialized with Set by the caller (the outer storage + // holds its mutex around every write). + WatermarkTs(key string) int64 +} + +// SetResult reports what one Set call actually changed: the values that won +// their LWW comparison and were upserted, and the ids of rows a watermark in +// the batch physically removed. Values silently skipped — older than the +// stored row, or covered by a watermark — appear in neither. +type SetResult struct { + Applied []KeyValue + DroppedIds []string } type storage struct { @@ -137,9 +150,17 @@ func (s *storage) IterateValues(ctx context.Context, iterFunc func(kv KeyValue) } func (s *storage) IteratePrefix(ctx context.Context, prefix string, iterFunc func(kv KeyValue) error) (err error) { + return s.iterateIdPrefix(ctx, prefix, func(doc anystore.Doc) error { + return iterFunc(s.keyValueFromDoc(doc)) + }) +} + +// iterateIdPrefix walks documents whose id starts with prefix, in id order: +// ids are sorted, so the matching rows form one contiguous run from the Gte +// seek, and the walk stops at the first id past it. +func (s *storage) iterateIdPrefix(ctx context.Context, prefix string, fn func(doc anystore.Doc) error) (err error) { filter := query.Key{Path: []string{"id"}, Filter: query.NewComp(query.CompOpGte, prefix)} - qry := s.collection.Find(filter).Sort("id") - iter, err := qry.Iter(ctx) + iter, err := s.collection.Find(filter).Sort("id").Iter(ctx) if err != nil { return } @@ -151,12 +172,11 @@ func (s *storage) IteratePrefix(ctx context.Context, prefix string, iterFunc fun if doc, err = iter.Doc(); err != nil { return } - if !strings.Contains(doc.Value().GetString("id"), prefix) { + if !strings.HasPrefix(doc.Value().GetString("id"), prefix) { break } - err := iterFunc(s.keyValueFromDoc(doc)) - if err != nil { - return err + if err = fn(doc); err != nil { + return } } return nil @@ -205,15 +225,12 @@ func (s *storage) init(ctx context.Context) (err error) { return } -func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { +func (s *storage) Set(ctx context.Context, values ...KeyValue) (result SetResult, err error) { tx, err := s.collection.WriteTx(ctx) if err != nil { return } - var ( - res updateResult - diffUpdated bool - ) + var res updateResult defer func() { if err == nil { err = tx.Commit() @@ -226,23 +243,19 @@ func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { s.watermarks[prefix] = t } } + return } - if err != nil && diffUpdated { - // The diff already contains this call's elements but the tx did not - // commit: undo the in-memory mutations so the diff never advertises - // heads the storage doesn't hold — peers would never re-send those - // values. Prior heads are restored in reverse so a duplicate id ends - // up at its genuine pre-call state; inserts are then dropped, and - // watermark-dropped rows re-added (the tx rollback restored their - // documents). - for i := len(res.prior) - 1; i >= 0; i-- { - s.diff.Set(res.prior[i]) - } - for _, id := range res.added { + // The diff already contains this call's mutations but the tx did not + // commit: restore every touched element to its pre-call state so the + // diff never advertises heads the storage doesn't hold — peers would + // never re-send those values. preState records exactly one snapshot + // per id, taken before its first mutation, so any interleaving of + // upserts and watermark drops within the batch undoes correctly. + for id, pre := range res.preState { + if pre == nil { _ = s.diff.RemoveId(id) - } - for _, el := range res.removed { - s.diff.Set(el) + } else { + s.diff.Set(*pre) } } }() @@ -255,41 +268,65 @@ func (s *storage) Set(ctx context.Context, values ...KeyValue) (err error) { for _, el := range res.removed { _ = s.diff.RemoveId(el.Id) } - diffUpdated = len(res.elements) > 0 || len(res.removed) > 0 + if len(res.preState) == 0 { + return SetResult{}, nil // nothing won LWW; skip the head update + } err = s.headStorage.UpdateEntry(ctx, headstorage.HeadsUpdate{ Id: s.storageName, Heads: []string{s.diff.Hash()}, }) - return + if err != nil { + return + } + return SetResult{Applied: res.applied, DroppedIds: res.droppedIds}, nil } -// updateResult carries one updateValues batch outcome: the new diff elements, -// what is needed to undo the diff on a failed tx (replaced elements' prior -// state, inserted ids, elements of watermark-dropped rows), and the watermarks -// to publish into s.watermarks once the tx commits. +func (s *storage) WatermarkTs(key string) int64 { + var maxTs int64 + for prefix, t := range s.watermarks { + if t > maxTs && strings.HasPrefix(key, prefix) { + maxTs = t + } + } + return maxTs +} + +// updateResult carries one updateValues batch outcome: the diff elements to +// advertise and withdraw, the per-id pre-call snapshots that undo the diff on +// a failed tx, the watermarks to publish into s.watermarks once the tx +// commits, and the applied/dropped sets reported to the caller. type updateResult struct { - elements []ldiff.Element - prior []ldiff.Element - added []string - removed []ldiff.Element + elements []ldiff.Element + removed []ldiff.Element + // preState maps every id mutated in this batch to its diff element + // before the call (nil = absent), recorded once at first mutation. + preState map[string]*ldiff.Element watermarks map[string]int64 + applied []KeyValue + droppedIds []string +} + +// recordPre snapshots an id's pre-call element once; later mutations of the +// same id within the batch keep the first (genuine) snapshot. +func (r *updateResult) recordPre(id string, el *ldiff.Element) { + if _, ok := r.preState[id]; !ok { + r.preState[id] = el + } } // covered reports whether a row loses to an applied or in-batch watermark: // its key falls under the prefix and it is strictly older. A watermark row // itself never loses to its own prefix (equal timestamps are not covered). func (s *storage) covered(res *updateResult, key string, timestampMicro int64) bool { - for prefix, t := range s.watermarks { - if timestampMicro < t && strings.HasPrefix(key, prefix) { - return true - } - } - for prefix, t := range res.watermarks { - if timestampMicro < t && strings.HasPrefix(key, prefix) { - return true + match := func(watermarks map[string]int64) bool { + for prefix, t := range watermarks { + if timestampMicro < t && strings.HasPrefix(key, prefix) { + return true + } } + return false } - return false + return match(s.watermarks) || match(res.watermarks) } // updateValues upserts the values that win their LWW comparison. A winning @@ -303,6 +340,7 @@ func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (res upd defer arenaPool.Put(arena) res.elements = make([]ldiff.Element, 0, len(values)) + res.preState = map[string]*ldiff.Element{} res.watermarks = map[string]int64{} var doc anystore.Doc for _, value := range values { @@ -318,9 +356,10 @@ func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (res upd if int64(doc.Value().GetFloat64("t")) >= value.TimestampMicro { continue } - res.prior = append(res.prior, anyEncToElement(doc.Value())) + el := anyEncToElement(doc.Value()) + res.recordPre(el.Id, &el) } else { - res.added = append(res.added, value.KeyPeerId) + res.recordPre(value.KeyPeerId, nil) } arena.Reset() val := value.AnyEnc(arena) @@ -328,6 +367,7 @@ func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (res upd return } res.elements = append(res.elements, anyEncToElement(val)) + res.applied = append(res.applied, value) if value.DeletePrefix != "" { if err = s.applyWatermark(ctx, &res, value); err != nil { return @@ -337,34 +377,30 @@ func (s *storage) updateValues(ctx context.Context, values ...KeyValue) (res upd return } -// applyWatermark physically deletes every stored row whose key starts with +// applyWatermark physically deletes every stored row whose KEY starts with // the watermark's prefix and is strictly older than it, the just-written // watermark row excepted by the timestamp comparison. Runs inside Set's tx; // diff elements of the dropped rows are removed by Set after it returns. func (s *storage) applyWatermark(ctx context.Context, res *updateResult, wm KeyValue) (err error) { - // KeyPeerId starts with Key, so a key-prefix scan is an id-prefix scan. - filter := query.Key{Path: []string{"id"}, Filter: query.NewComp(query.CompOpGte, wm.DeletePrefix)} - iter, err := s.collection.Find(filter).Sort("id").Iter(ctx) - if err != nil { - return err - } var dropIds []string - var doc anystore.Doc - for iter.Next() { - if doc, err = iter.Doc(); err != nil { - _ = iter.Close() - return err - } - if !strings.HasPrefix(doc.Value().GetString("id"), wm.DeletePrefix) { - break + // A key under the prefix always puts its ids (key+"-"+peerId) inside the + // id-prefix run, so the scan is bounded — but not vice versa: an id can + // match past its key's end ("cnt-…" peer ids under prefix "cnt-1" belong + // to key "cnt"), so the key itself must be checked per document. + err = s.iterateIdPrefix(ctx, wm.DeletePrefix, func(doc anystore.Doc) error { + if !strings.HasPrefix(doc.Value().GetString("k"), wm.DeletePrefix) { + return nil } if int64(doc.Value().GetFloat64("t")) >= wm.TimestampMicro { - continue + return nil } - res.removed = append(res.removed, anyEncToElement(doc.Value())) - dropIds = append(dropIds, doc.Value().GetString("id")) - } - if err = iter.Close(); err != nil { + el := anyEncToElement(doc.Value()) + res.recordPre(el.Id, &el) + res.removed = append(res.removed, el) + dropIds = append(dropIds, el.Id) + return nil + }) + if err != nil { return err } for _, id := range dropIds { @@ -372,6 +408,7 @@ func (s *storage) applyWatermark(ctx context.Context, res *updateResult, wm KeyV return err } } + res.droppedIds = append(res.droppedIds, dropIds...) if wm.TimestampMicro > res.watermarks[wm.DeletePrefix] { res.watermarks[wm.DeletePrefix] = wm.TimestampMicro } diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go index c3337637c..612272072 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/keyvaluestorage_test.go @@ -31,6 +31,13 @@ func (f *failingHeadStorage) UpdateEntry(ctx context.Context, update headstorage return f.HeadStorage.UpdateEntry(ctx, update) } +func mustSet(t *testing.T, st innerstorage.KeyValueStorage, kvs ...innerstorage.KeyValue) innerstorage.SetResult { + t.Helper() + res, err := st.Set(ctx, kvs...) + require.NoError(t, err) + return res +} + func newTestStorage(t *testing.T) innerstorage.KeyValueStorage { storage, _ := newTestStorageWithFailingHeads(t) return storage @@ -55,12 +62,13 @@ func newTestStorageWithFailingHeads(t *testing.T) (innerstorage.KeyValueStorage, // storage until restart). func TestSetFailureKeepsDiffConsistent(t *testing.T) { storage, failingHeads := newTestStorageWithFailingHeads(t) - require.NoError(t, storage.Set(ctx, testKeyValue(0))) + mustSet(t, storage, testKeyValue(0)) hashBefore := storage.Diff().Hash() failingHeads.fail = true kv := testKeyValue(1) - require.Error(t, storage.Set(ctx, kv)) + _, setErr := storage.Set(ctx, kv) + require.Error(t, setErr) failingHeads.fail = false _, err := storage.Diff().Element(kv.KeyPeerId) @@ -112,7 +120,7 @@ func TestIterateValuesKeyRowsAdjacent(t *testing.T) { kv.KeyPeerId = kv.Key + "-" + kv.PeerId batch = append(batch, kv) } - require.NoError(t, storage.Set(ctx, batch...)) + mustSet(t, storage, batch...) } var ids, keysSeen []string @@ -145,7 +153,7 @@ func TestIterateValuesReturnsOwnedMemory(t *testing.T) { for i := 0; i < 3; i++ { kv := testKeyValue(i) originals[kv.KeyPeerId] = kv - require.NoError(t, storage.Set(ctx, kv)) + mustSet(t, storage, kv) } var collected []innerstorage.KeyValue @@ -204,17 +212,17 @@ func storedIds(t *testing.T, storage innerstorage.KeyValueStorage) []string { // newer rows and rows outside the prefix survive. func TestWatermarkDropsOlderRows(t *testing.T) { storage := newTestStorage(t) - require.NoError(t, storage.Set(ctx, + mustSet(t, storage, rowKV("read/sp1/obj1", "peerA", 10), rowKV("read/sp1/obj1", "peerB", 11), rowKV("read/sp1/obj2", "peerA", 12), rowKV("read/sp1/obj3", "peerA", 200), // newer than the watermark rowKV("read/sp2/obj1", "peerA", 13), // outside the prefix - )) + ) hashBefore := storage.Diff().Hash() wm := wmKV("read/sp1/", "peerA", 100) - require.NoError(t, storage.Set(ctx, wm)) + mustSet(t, storage, wm) require.Equal(t, []string{ "read/sp1/-peerA", // the watermark row itself @@ -234,13 +242,13 @@ func TestWatermarkDropsOlderRows(t *testing.T) { // applied watermark are discarded on arrival; equal-or-newer rows apply. func TestWatermarkRejectsLateArrivals(t *testing.T) { storage := newTestStorage(t) - require.NoError(t, storage.Set(ctx, wmKV("read/sp1/", "peerA", 100))) + mustSet(t, storage, wmKV("read/sp1/", "peerA", 100)) - require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj1", "peerB", 99))) + mustSet(t, storage, rowKV("read/sp1/obj1", "peerB", 99)) _, err := storage.GetKeyPeerId(ctx, "read/sp1/obj1-peerB") require.ErrorIs(t, err, anystore.ErrDocNotFound, "older row must be rejected") - require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj2", "peerB", 100))) + mustSet(t, storage, rowKV("read/sp1/obj2", "peerB", 100)) _, err = storage.GetKeyPeerId(ctx, "read/sp1/obj2-peerB") require.NoError(t, err, "equal-timestamp row is not covered") } @@ -255,11 +263,11 @@ func TestWatermarkSurvivesReopen(t *testing.T) { require.NoError(t, err) storage, err := innerstorage.New(ctx, "kv.test", heads, db) require.NoError(t, err) - require.NoError(t, storage.Set(ctx, wmKV("read/sp1/", "peerA", 100))) + mustSet(t, storage, wmKV("read/sp1/", "peerA", 100)) reopened, err := innerstorage.New(ctx, "kv.test", heads, db) require.NoError(t, err) - require.NoError(t, reopened.Set(ctx, rowKV("read/sp1/obj1", "peerB", 99))) + mustSet(t, reopened, rowKV("read/sp1/obj1", "peerB", 99)) _, err = reopened.GetKeyPeerId(ctx, "read/sp1/obj1-peerB") require.ErrorIs(t, err, anystore.ErrDocNotFound, "watermark must reject after reopen") } @@ -273,7 +281,7 @@ func TestWatermarkBatchOrderIndependent(t *testing.T) { } { t.Run(name, func(t *testing.T) { storage := newTestStorage(t) - require.NoError(t, storage.Set(ctx, batch...)) + mustSet(t, storage, batch...) require.Equal(t, []string{"read/sp1/-peerA"}, storedIds(t, storage)) }) } @@ -285,12 +293,13 @@ func TestWatermarkBatchOrderIndependent(t *testing.T) { func TestWatermarkFailedTxRestoresDroppedElements(t *testing.T) { storage, failingHeads := newTestStorageWithFailingHeads(t) row := rowKV("read/sp1/obj1", "peerB", 50) - require.NoError(t, storage.Set(ctx, row)) + mustSet(t, storage, row) hashBefore := storage.Diff().Hash() failingHeads.fail = true wm := wmKV("read/sp1/", "peerA", 100) - require.Error(t, storage.Set(ctx, wm)) + _, setErr := storage.Set(ctx, wm) + require.Error(t, setErr) failingHeads.fail = false _, err := storage.Diff().Element(row.KeyPeerId) @@ -303,7 +312,67 @@ func TestWatermarkFailedTxRestoresDroppedElements(t *testing.T) { // The rolled-back watermark must not keep rejecting: a fresh old row // still applies. - require.NoError(t, storage.Set(ctx, rowKV("read/sp1/obj2", "peerB", 60))) + mustSet(t, storage, rowKV("read/sp1/obj2", "peerB", 60)) _, err = storage.GetKeyPeerId(ctx, "read/sp1/obj2-peerB") require.NoError(t, err) } + +// TestWatermarkKeyBoundary: the drop matches on the KEY, not the id — a +// prefix that extends past a full key into the "-"+peerId separator must not +// delete that key's rows, and must not reject its later arrivals. +func TestWatermarkKeyBoundary(t *testing.T) { + storage := newTestStorage(t) + mustSet(t, storage, + rowKV("cnt", "1AB", 10), // id "cnt-1AB" matches prefix "cnt-1"; key "cnt" does not + rowKV("cnt-10", "peerA", 11), // key genuinely under the prefix + ) + mustSet(t, storage, wmKV("cnt-1", "peerA", 100)) + + require.Equal(t, []string{"cnt-1-peerA", "cnt-1AB"}, storedIds(t, storage), + "key outside the prefix survives; key under it drops") + mustSet(t, storage, rowKV("cnt", "1CD", 50)) + _, err := storage.GetKeyPeerId(ctx, "cnt-1CD") + require.NoError(t, err, "late arrival for a key outside the prefix must apply") +} + +// TestWatermarkFailedTxSameBatch: when a batch writes a row and a watermark +// that drops it and the tx then fails, the undo must leave no trace — a +// leftover advertised element would make the LWW pre-check discard the +// genuine row forever. +func TestWatermarkFailedTxSameBatch(t *testing.T) { + storage, failingHeads := newTestStorageWithFailingHeads(t) + hashBefore := storage.Diff().Hash() + failingHeads.fail = true + row := rowKV("read/sp1/obj1", "peerB", 50) + wm := wmKV("read/sp1/", "peerA", 100) + _, setErr := storage.Set(ctx, row, wm) + require.Error(t, setErr) + failingHeads.fail = false + + _, err := storage.Diff().Element(row.KeyPeerId) + require.ErrorIs(t, err, ldiff.ErrElementNotFound, "rolled-back row must not be advertised") + _, err = storage.Diff().Element(wm.KeyPeerId) + require.ErrorIs(t, err, ldiff.ErrElementNotFound, "rolled-back watermark must not be advertised") + require.Equal(t, hashBefore, storage.Diff().Hash()) + + res := mustSet(t, storage, row) + require.Len(t, res.Applied, 1, "the identical row must be storable after the rollback") + _, err = storage.GetKeyPeerId(ctx, row.KeyPeerId) + require.NoError(t, err) +} + +// TestSetResultReportsAppliedAndDropped pins the SetResult contract: applied +// carries only LWW winners, droppedIds the watermark-removed rows. +func TestSetResultReportsAppliedAndDropped(t *testing.T) { + storage := newTestStorage(t) + res := mustSet(t, storage, rowKV("read/sp1/obj1", "peerA", 10)) + require.Len(t, res.Applied, 1) + require.Empty(t, res.DroppedIds) + + res = mustSet(t, storage, rowKV("read/sp1/obj1", "peerA", 5)) // LWW loser + require.Empty(t, res.Applied) + + res = mustSet(t, storage, wmKV("read/sp1/", "peerB", 100)) + require.Len(t, res.Applied, 1, "the watermark row itself applies") + require.Equal(t, []string{"read/sp1/obj1-peerA"}, res.DroppedIds) +} diff --git a/commonspace/object/keyvalue/keyvaluestorage/storage.go b/commonspace/object/keyvalue/keyvaluestorage/storage.go index 1a58d195e..29acf30d1 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/storage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/storage.go @@ -4,6 +4,7 @@ package keyvaluestorage import ( "context" "encoding/binary" + "errors" "fmt" "sync" "time" @@ -25,6 +26,13 @@ import ( var log = logger.NewNamed("common.keyvalue.keyvaluestorage") +var ( + // ErrCoveredByWatermark: the write's key sits under a deletion watermark + // newer than the write's timestamp, so every replica would discard it. + ErrCoveredByWatermark = errors.New("key is covered by a newer deletion watermark") + ErrEmptyDeletePrefix = errors.New("empty delete prefix") +) + const IndexerCName = "common.keyvalue.indexer" type Indexer interface { @@ -32,6 +40,14 @@ type Indexer interface { Index(decryptor Decryptor, keyValue ...innerstorage.KeyValue) error } +// DeletionAwareIndexer is implemented by indexers that mirror physical +// deletions: RemoveIndex receives the ids of rows a watermark dropped, so the +// downstream index does not keep resolving keys the storage no longer holds. +type DeletionAwareIndexer interface { + Indexer + RemoveIndex(keyPeerIds ...string) error +} + type Decryptor = func(kv innerstorage.KeyValue) (value []byte, err error) type NoOpIndexer struct{} @@ -129,6 +145,9 @@ func (s *storage) Set(ctx context.Context, key string, value []byte) error { } func (s *storage) DeletePrefix(ctx context.Context, prefix string) error { + if prefix == "" { + return ErrEmptyDeletePrefix + } s.mx.Lock() defer s.mx.Unlock() // Owner-only: a watermark removes rows written by every peer, not just @@ -173,6 +192,12 @@ func (s *storage) signAndStore(ctx context.Context, key string, value []byte, de return err } timestampMicro := time.Now().UnixMicro() + // Fail loudly instead of writing a row every replica (this one included) + // would silently discard: without this, Set would return success and the + // value would never be readable anywhere. + if wmTs := s.inner.WatermarkTs(key); wmTs > timestampMicro { + return ErrCoveredByWatermark + } inner := spacesyncproto.StoreKeyInner{ Peer: protoPeerKey, Identity: protoIdentityKey, @@ -210,10 +235,15 @@ func (s *storage) signAndStore(ctx context.Context, key string, value []byte, de IdentitySignature: identitySig, }, } - err = s.inner.Set(ctx, keyValue) + res, err := s.inner.Set(ctx, keyValue) if err != nil { return err } + s.removeIndexes(res.DroppedIds) + if len(res.Applied) == 0 { + // Lost LWW to an already-stored newer own row; nothing changed. + return nil + } if deletePrefix == "" { indexErr := s.indexer.Index(s.decrypt, keyValue) if indexErr != nil { @@ -227,6 +257,20 @@ func (s *storage) signAndStore(ctx context.Context, key string, value []byte, de return nil } +// removeIndexes mirrors watermark drops into the indexer when it opts in. +func (s *storage) removeIndexes(droppedIds []string) { + if len(droppedIds) == 0 { + return + } + indexer, ok := s.indexer.(DeletionAwareIndexer) + if !ok { + return + } + if err := indexer.RemoveIndex(droppedIds...); err != nil { + log.Warn("failed to remove index for dropped keys", zap.Error(err)) + } +} + func (s *storage) SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreKeyValue) (err error) { if len(keyValue) == 0 { return nil @@ -281,17 +325,24 @@ func (s *storage) SetRaw(ctx context.Context, keyValue ...*spacesyncproto.StoreK if len(keyValues) == 0 { return nil } - err = s.inner.Set(ctx, keyValues...) + res, err := s.inner.Set(ctx, keyValues...) if err != nil { return err } - sendErr := s.syncClient.Broadcast(ctx, s.storageId, keyValues...) + s.removeIndexes(res.DroppedIds) + // Broadcast and index only what actually applied: values the store + // rejected (LWW losers, watermark-covered rows) must not propagate + // further — indexing them would resurrect deleted data downstream. + if len(res.Applied) == 0 { + return nil + } + sendErr := s.syncClient.Broadcast(ctx, s.storageId, res.Applied...) if sendErr != nil { log.Warn("failed to send key values", zap.Error(sendErr)) } // Watermarks carry no payload to index; their effect (dropped rows) is // already applied. - indexable := slice.DiscardFromSlice(keyValues, func(value innerstorage.KeyValue) bool { + indexable := slice.DiscardFromSlice(res.Applied, func(value innerstorage.KeyValue) bool { return value.DeletePrefix != "" }) if len(indexable) > 0 { From 36d4e379683e16c6930eac5ecb5bd27365b1b9e0 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 11 Aug 2026 18:49:53 +0200 Subject: [PATCH 3/4] keyvalue: watermark as a typed StoreDeletePrefix sub-message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deletion operation is its own message inside the signed StoreKeyInner (inner.delete != nil) instead of a bare string field: the schema states what the field does, the operation type cannot be forged (an unsigned outer discriminator could be), and future deletion options have a typed home. Wire mechanics unchanged — the watermark stays an ordinary synced row, which anti-entropy and p2p sync require. SYN-145 --- commonspace/object/keyvalue/keyvalue_test.go | 2 +- .../keyvaluestorage/innerstorage/element.go | 2 +- .../keyvalue/keyvaluestorage/storage.go | 4 +- .../spacesyncproto/protos/spacesync.proto | 20 ++- commonspace/spacesyncproto/spacesync.pb.go | 136 ++++++++++----- .../spacesyncproto/spacesync_vtproto.pb.go | 158 +++++++++++++++++- 6 files changed, 264 insertions(+), 58 deletions(-) diff --git a/commonspace/object/keyvalue/keyvalue_test.go b/commonspace/object/keyvalue/keyvalue_test.go index 6469c9e8b..e68750a5e 100644 --- a/commonspace/object/keyvalue/keyvalue_test.go +++ b/commonspace/object/keyvalue/keyvalue_test.go @@ -465,7 +465,7 @@ func rawWatermark(t *testing.T, keys *accountdata.AccountKeys, prefix string, ts TimestampMicro: ts, AclHeadId: "acl-head", Key: prefix, - DeletePrefix: prefix, + Delete: &spacesyncproto.StoreDeletePrefix{Prefix: prefix}, } innerBytes, err := inner.MarshalVT() require.NoError(t, err) diff --git a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go index b5bd72905..7151b7d77 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go +++ b/commonspace/object/keyvalue/keyvaluestorage/innerstorage/element.go @@ -65,7 +65,7 @@ func KeyValueFromProto(proto *spacesyncproto.StoreKeyValue, verify bool) (kv Key kv.PeerId = peerId.PeerId() kv.Key = innerValue.Key kv.AclId = innerValue.AclHeadId - kv.DeletePrefix = innerValue.DeletePrefix + kv.DeletePrefix = innerValue.GetDelete().GetPrefix() // The id must be derived from the signed payload: an unchecked KeyPeerId // would let any writer occupy (and overwrite) another peer's row. if kv.KeyPeerId != kv.Key+"-"+kv.PeerId { diff --git a/commonspace/object/keyvalue/keyvaluestorage/storage.go b/commonspace/object/keyvalue/keyvaluestorage/storage.go index 29acf30d1..45f1b97cc 100644 --- a/commonspace/object/keyvalue/keyvaluestorage/storage.go +++ b/commonspace/object/keyvalue/keyvaluestorage/storage.go @@ -205,7 +205,9 @@ func (s *storage) signAndStore(ctx context.Context, key string, value []byte, de TimestampMicro: timestampMicro, AclHeadId: headId, Key: key, - DeletePrefix: deletePrefix, + } + if deletePrefix != "" { + inner.Delete = &spacesyncproto.StoreDeletePrefix{Prefix: deletePrefix} } innerBytes, err := inner.MarshalVT() if err != nil { diff --git a/commonspace/spacesyncproto/protos/spacesync.proto b/commonspace/spacesyncproto/protos/spacesync.proto index af20330c7..393479945 100644 --- a/commonspace/spacesyncproto/protos/spacesync.proto +++ b/commonspace/spacesyncproto/protos/spacesync.proto @@ -262,12 +262,20 @@ message StoreKeyInner { int64 timestampMicro = 4; string aclHeadId = 5; string key = 6; - // deletePrefix marks the row as a deletion watermark: applying it - // physically removes every stored row whose key starts with the - // prefix and whose timestamp is older than timestampMicro, and - // rejects such rows arriving later. The watermark row itself is the - // only retained state. key mirrors the prefix; value stays empty. - string deletePrefix = 7; + // delete turns this row into a deletion watermark instead of a value + // (see StoreDeletePrefix). key mirrors the prefix; value stays empty. + StoreDeletePrefix delete = 7; +} + +// StoreDeletePrefix is the deletion watermark operation: applying the row +// carrying it physically removes every stored row whose key starts with the +// prefix and whose timestamp is older than the watermark's, and rejects such +// rows arriving later. The watermark row itself is the only retained state. +// It lives inside the signed StoreKeyInner so the operation type cannot be +// forged, and rides the ordinary element sync so it reaches every replica — +// including devices restoring from old snapshots. +message StoreDeletePrefix { + string prefix = 1; } message StorageHeader { diff --git a/commonspace/spacesyncproto/spacesync.pb.go b/commonspace/spacesyncproto/spacesync.pb.go index 2ecde95ec..d4bf3805a 100644 --- a/commonspace/spacesyncproto/spacesync.pb.go +++ b/commonspace/spacesyncproto/spacesync.pb.go @@ -2075,12 +2075,9 @@ type StoreKeyInner struct { TimestampMicro int64 `protobuf:"varint,4,opt,name=timestampMicro,proto3" json:"timestampMicro,omitempty"` AclHeadId string `protobuf:"bytes,5,opt,name=aclHeadId,proto3" json:"aclHeadId,omitempty"` Key string `protobuf:"bytes,6,opt,name=key,proto3" json:"key,omitempty"` - // deletePrefix marks the row as a deletion watermark: applying it - // physically removes every stored row whose key starts with the - // prefix and whose timestamp is older than timestampMicro, and - // rejects such rows arriving later. The watermark row itself is the - // only retained state. key mirrors the prefix; value stays empty. - DeletePrefix string `protobuf:"bytes,7,opt,name=deletePrefix,proto3" json:"deletePrefix,omitempty"` + // delete turns this row into a deletion watermark instead of a value + // (see StoreDeletePrefix). key mirrors the prefix; value stays empty. + Delete *StoreDeletePrefix `protobuf:"bytes,7,opt,name=delete,proto3" json:"delete,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2157,9 +2154,60 @@ func (x *StoreKeyInner) GetKey() string { return "" } -func (x *StoreKeyInner) GetDeletePrefix() string { +func (x *StoreKeyInner) GetDelete() *StoreDeletePrefix { if x != nil { - return x.DeletePrefix + return x.Delete + } + return nil +} + +// StoreDeletePrefix is the deletion watermark operation: applying the row +// carrying it physically removes every stored row whose key starts with the +// prefix and whose timestamp is older than the watermark's, and rejects such +// rows arriving later. The watermark row itself is the only retained state. +// It lives inside the signed StoreKeyInner so the operation type cannot be +// forged, and rides the ordinary element sync so it reaches every replica — +// including devices restoring from old snapshots. +type StoreDeletePrefix struct { + state protoimpl.MessageState `protogen:"open.v1"` + Prefix string `protobuf:"bytes,1,opt,name=prefix,proto3" json:"prefix,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StoreDeletePrefix) Reset() { + *x = StoreDeletePrefix{} + mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[31] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StoreDeletePrefix) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StoreDeletePrefix) ProtoMessage() {} + +func (x *StoreDeletePrefix) ProtoReflect() protoreflect.Message { + mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[31] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StoreDeletePrefix.ProtoReflect.Descriptor instead. +func (*StoreDeletePrefix) Descriptor() ([]byte, []int) { + return file_commonspace_spacesyncproto_protos_spacesync_proto_rawDescGZIP(), []int{31} +} + +func (x *StoreDeletePrefix) GetPrefix() string { + if x != nil { + return x.Prefix } return "" } @@ -2174,7 +2222,7 @@ type StorageHeader struct { func (x *StorageHeader) Reset() { *x = StorageHeader{} - mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[31] + mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[32] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2186,7 +2234,7 @@ func (x *StorageHeader) String() string { func (*StorageHeader) ProtoMessage() {} func (x *StorageHeader) ProtoReflect() protoreflect.Message { - mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[31] + mi := &file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes[32] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2199,7 +2247,7 @@ func (x *StorageHeader) ProtoReflect() protoreflect.Message { // Deprecated: Use StorageHeader.ProtoReflect.Descriptor instead. func (*StorageHeader) Descriptor() ([]byte, []int) { - return file_commonspace_spacesyncproto_protos_spacesync_proto_rawDescGZIP(), []int{31} + return file_commonspace_spacesyncproto_protos_spacesync_proto_rawDescGZIP(), []int{32} } func (x *StorageHeader) GetSpaceId() string { @@ -2339,15 +2387,17 @@ const file_commonspace_spacesyncproto_protos_spacesync_proto_rawDesc = "" + "\rpeerSignature\x18\x04 \x01(\fR\rpeerSignature\x12\x18\n" + "\aspaceId\x18\x05 \x01(\tR\aspaceId\"H\n" + "\x0eStoreKeyValues\x126\n" + - "\tkeyValues\x18\x01 \x03(\v2\x18.spacesync.StoreKeyValueR\tkeyValues\"\xd1\x01\n" + + "\tkeyValues\x18\x01 \x03(\v2\x18.spacesync.StoreKeyValueR\tkeyValues\"\xe3\x01\n" + "\rStoreKeyInner\x12\x12\n" + "\x04peer\x18\x01 \x01(\fR\x04peer\x12\x1a\n" + "\bidentity\x18\x02 \x01(\fR\bidentity\x12\x14\n" + "\x05value\x18\x03 \x01(\fR\x05value\x12&\n" + "\x0etimestampMicro\x18\x04 \x01(\x03R\x0etimestampMicro\x12\x1c\n" + "\taclHeadId\x18\x05 \x01(\tR\taclHeadId\x12\x10\n" + - "\x03key\x18\x06 \x01(\tR\x03key\x12\"\n" + - "\fdeletePrefix\x18\a \x01(\tR\fdeletePrefix\"K\n" + + "\x03key\x18\x06 \x01(\tR\x03key\x124\n" + + "\x06delete\x18\a \x01(\v2\x1c.spacesync.StoreDeletePrefixR\x06delete\"+\n" + + "\x11StoreDeletePrefix\x12\x16\n" + + "\x06prefix\x18\x01 \x01(\tR\x06prefix\"K\n" + "\rStorageHeader\x12\x18\n" + "\aspaceId\x18\x01 \x01(\tR\aspaceId\x12 \n" + "\vstorageName\x18\x02 \x01(\tR\vstorageName*\xee\x01\n" + @@ -2409,7 +2459,7 @@ func file_commonspace_spacesyncproto_protos_spacesync_proto_rawDescGZIP() []byte } var file_commonspace_spacesyncproto_protos_spacesync_proto_enumTypes = make([]protoimpl.EnumInfo, 6) -var file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes = make([]protoimpl.MessageInfo, 32) +var file_commonspace_spacesyncproto_protos_spacesync_proto_msgTypes = make([]protoimpl.MessageInfo, 33) var file_commonspace_spacesyncproto_protos_spacesync_proto_goTypes = []any{ (ErrCodes)(0), // 0: spacesync.ErrCodes (SpaceHeaderVersion)(0), // 1: spacesync.SpaceHeaderVersion @@ -2448,7 +2498,8 @@ var file_commonspace_spacesyncproto_protos_spacesync_proto_goTypes = []any{ (*StoreKeyValue)(nil), // 34: spacesync.StoreKeyValue (*StoreKeyValues)(nil), // 35: spacesync.StoreKeyValues (*StoreKeyInner)(nil), // 36: spacesync.StoreKeyInner - (*StorageHeader)(nil), // 37: spacesync.StorageHeader + (*StoreDeletePrefix)(nil), // 37: spacesync.StoreDeletePrefix + (*StorageHeader)(nil), // 38: spacesync.StorageHeader } var file_commonspace_spacesyncproto_protos_spacesync_proto_depIdxs = []int32{ 8, // 0: spacesync.HeadSyncResult.elements:type_name -> spacesync.HeadSyncResultElement @@ -2471,31 +2522,32 @@ var file_commonspace_spacesyncproto_protos_spacesync_proto_depIdxs = []int32{ 6, // 17: spacesync.StoreDiffRequest.ranges:type_name -> spacesync.HeadSyncRange 7, // 18: spacesync.StoreDiffResponse.results:type_name -> spacesync.HeadSyncResult 34, // 19: spacesync.StoreKeyValues.keyValues:type_name -> spacesync.StoreKeyValue - 9, // 20: spacesync.SpaceSync.HeadSync:input_type -> spacesync.HeadSyncRequest - 32, // 21: spacesync.SpaceSync.StoreDiff:input_type -> spacesync.StoreDiffRequest - 34, // 22: spacesync.SpaceSync.StoreElements:input_type -> spacesync.StoreKeyValue - 12, // 23: spacesync.SpaceSync.SpacePush:input_type -> spacesync.SpacePushRequest - 14, // 24: spacesync.SpaceSync.SpacePull:input_type -> spacesync.SpacePullRequest - 11, // 25: spacesync.SpaceSync.ObjectSyncStream:input_type -> spacesync.ObjectSyncMessage - 11, // 26: spacesync.SpaceSync.ObjectSync:input_type -> spacesync.ObjectSyncMessage - 11, // 27: spacesync.SpaceSync.ObjectSyncRequestStream:input_type -> spacesync.ObjectSyncMessage - 28, // 28: spacesync.SpaceSync.AclAddRecord:input_type -> spacesync.AclAddRecordRequest - 30, // 29: spacesync.SpaceSync.AclGetRecords:input_type -> spacesync.AclGetRecordsRequest - 10, // 30: spacesync.SpaceSync.HeadSync:output_type -> spacesync.HeadSyncResponse - 33, // 31: spacesync.SpaceSync.StoreDiff:output_type -> spacesync.StoreDiffResponse - 34, // 32: spacesync.SpaceSync.StoreElements:output_type -> spacesync.StoreKeyValue - 13, // 33: spacesync.SpaceSync.SpacePush:output_type -> spacesync.SpacePushResponse - 15, // 34: spacesync.SpaceSync.SpacePull:output_type -> spacesync.SpacePullResponse - 11, // 35: spacesync.SpaceSync.ObjectSyncStream:output_type -> spacesync.ObjectSyncMessage - 11, // 36: spacesync.SpaceSync.ObjectSync:output_type -> spacesync.ObjectSyncMessage - 11, // 37: spacesync.SpaceSync.ObjectSyncRequestStream:output_type -> spacesync.ObjectSyncMessage - 29, // 38: spacesync.SpaceSync.AclAddRecord:output_type -> spacesync.AclAddRecordResponse - 31, // 39: spacesync.SpaceSync.AclGetRecords:output_type -> spacesync.AclGetRecordsResponse - 30, // [30:40] is the sub-list for method output_type - 20, // [20:30] is the sub-list for method input_type - 20, // [20:20] is the sub-list for extension type_name - 20, // [20:20] is the sub-list for extension extendee - 0, // [0:20] is the sub-list for field type_name + 37, // 20: spacesync.StoreKeyInner.delete:type_name -> spacesync.StoreDeletePrefix + 9, // 21: spacesync.SpaceSync.HeadSync:input_type -> spacesync.HeadSyncRequest + 32, // 22: spacesync.SpaceSync.StoreDiff:input_type -> spacesync.StoreDiffRequest + 34, // 23: spacesync.SpaceSync.StoreElements:input_type -> spacesync.StoreKeyValue + 12, // 24: spacesync.SpaceSync.SpacePush:input_type -> spacesync.SpacePushRequest + 14, // 25: spacesync.SpaceSync.SpacePull:input_type -> spacesync.SpacePullRequest + 11, // 26: spacesync.SpaceSync.ObjectSyncStream:input_type -> spacesync.ObjectSyncMessage + 11, // 27: spacesync.SpaceSync.ObjectSync:input_type -> spacesync.ObjectSyncMessage + 11, // 28: spacesync.SpaceSync.ObjectSyncRequestStream:input_type -> spacesync.ObjectSyncMessage + 28, // 29: spacesync.SpaceSync.AclAddRecord:input_type -> spacesync.AclAddRecordRequest + 30, // 30: spacesync.SpaceSync.AclGetRecords:input_type -> spacesync.AclGetRecordsRequest + 10, // 31: spacesync.SpaceSync.HeadSync:output_type -> spacesync.HeadSyncResponse + 33, // 32: spacesync.SpaceSync.StoreDiff:output_type -> spacesync.StoreDiffResponse + 34, // 33: spacesync.SpaceSync.StoreElements:output_type -> spacesync.StoreKeyValue + 13, // 34: spacesync.SpaceSync.SpacePush:output_type -> spacesync.SpacePushResponse + 15, // 35: spacesync.SpaceSync.SpacePull:output_type -> spacesync.SpacePullResponse + 11, // 36: spacesync.SpaceSync.ObjectSyncStream:output_type -> spacesync.ObjectSyncMessage + 11, // 37: spacesync.SpaceSync.ObjectSync:output_type -> spacesync.ObjectSyncMessage + 11, // 38: spacesync.SpaceSync.ObjectSyncRequestStream:output_type -> spacesync.ObjectSyncMessage + 29, // 39: spacesync.SpaceSync.AclAddRecord:output_type -> spacesync.AclAddRecordResponse + 31, // 40: spacesync.SpaceSync.AclGetRecords:output_type -> spacesync.AclGetRecordsResponse + 31, // [31:41] is the sub-list for method output_type + 21, // [21:31] is the sub-list for method input_type + 21, // [21:21] is the sub-list for extension type_name + 21, // [21:21] is the sub-list for extension extendee + 0, // [0:21] is the sub-list for field type_name } func init() { file_commonspace_spacesyncproto_protos_spacesync_proto_init() } @@ -2513,7 +2565,7 @@ func file_commonspace_spacesyncproto_protos_spacesync_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_commonspace_spacesyncproto_protos_spacesync_proto_rawDesc), len(file_commonspace_spacesyncproto_protos_spacesync_proto_rawDesc)), NumEnums: 6, - NumMessages: 32, + NumMessages: 33, NumExtensions: 0, NumServices: 1, }, diff --git a/commonspace/spacesyncproto/spacesync_vtproto.pb.go b/commonspace/spacesyncproto/spacesync_vtproto.pb.go index c089704fb..4a3123ca4 100644 --- a/commonspace/spacesyncproto/spacesync_vtproto.pb.go +++ b/commonspace/spacesyncproto/spacesync_vtproto.pb.go @@ -1628,10 +1628,13 @@ func (m *StoreKeyInner) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } - if len(m.DeletePrefix) > 0 { - i -= len(m.DeletePrefix) - copy(dAtA[i:], m.DeletePrefix) - i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.DeletePrefix))) + if m.Delete != nil { + size, err := m.Delete.MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) i-- dAtA[i] = 0x3a } @@ -1678,6 +1681,46 @@ func (m *StoreKeyInner) MarshalToSizedBufferVT(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *StoreDeletePrefix) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *StoreDeletePrefix) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *StoreDeletePrefix) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + if len(m.Prefix) > 0 { + i -= len(m.Prefix) + copy(dAtA[i:], m.Prefix) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.Prefix))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + func (m *StorageHeader) MarshalVT() (dAtA []byte, err error) { if m == nil { return nil, nil @@ -2378,7 +2421,21 @@ func (m *StoreKeyInner) SizeVT() (n int) { if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } - l = len(m.DeletePrefix) + if m.Delete != nil { + l = m.Delete.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + n += len(m.unknownFields) + return n +} + +func (m *StoreDeletePrefix) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Prefix) if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } @@ -6438,7 +6495,94 @@ func (m *StoreKeyInner) UnmarshalVT(dAtA []byte) error { iNdEx = postIndex case 7: if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field DeletePrefix", wireType) + return fmt.Errorf("proto: wrong wireType = %d for field Delete", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Delete == nil { + m.Delete = &StoreDeletePrefix{} + } + if err := m.Delete.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *StoreDeletePrefix) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: StoreDeletePrefix: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: StoreDeletePrefix: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Prefix", wireType) } var stringLen uint64 for shift := uint(0); ; shift += 7 { @@ -6466,7 +6610,7 @@ func (m *StoreKeyInner) UnmarshalVT(dAtA []byte) error { if postIndex > l { return io.ErrUnexpectedEOF } - m.DeletePrefix = string(dAtA[iNdEx:postIndex]) + m.Prefix = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex default: iNdEx = preIndex From 7ea133451e6f63e404b2802bd301e13b7f2768ef Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 11 Aug 2026 18:58:18 +0200 Subject: [PATCH 4/4] keyvalue: offline-peer watermark scenarios Stale unsynced rows drop on both writer and hub once the watermark arrives; rows written after the deletion win by LWW and propagate until a re-issued delete removes them everywhere; unsynced rows outside the prefix sync normally. SYN-145 --- commonspace/object/keyvalue/keyvalue_test.go | 79 ++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/commonspace/object/keyvalue/keyvalue_test.go b/commonspace/object/keyvalue/keyvalue_test.go index e68750a5e..78e2a898a 100644 --- a/commonspace/object/keyvalue/keyvalue_test.go +++ b/commonspace/object/keyvalue/keyvalue_test.go @@ -686,3 +686,82 @@ func TestIndexerDeletionFlow(t *testing.T) { indexed, _ = idx.counts() require.Equal(t, 2, indexed, "rejected rows must not be indexed") } + +// attachPeer wires fx to hub's rpc server and returns the peer fx dials. +func attachPeer(t *testing.T, fx *fixture, fxKeys *accountdata.AccountKeys, hub *fixture, hubKeys *accountdata.AccountKeys) peer.Peer { + hubConn, fxConn := rpctest.MultiConnPair(fxKeys.PeerId, hubKeys.PeerId) + hubPeer, err := peer.NewPeer(hubConn, fx.server) + require.NoError(t, err) + _, err = peer.NewPeer(fxConn, hub.server) + require.NoError(t, err) + return hubPeer +} + +// TestDeletePrefix_OfflinePeers covers devices that were offline across a +// prefix deletion: +// - stale unsynced rows (older than the watermark) are rejected by the hub +// and dropped by their writer once the watermark reaches it; +// - rows written after the deletion (newer than the watermark) win by LWW +// and propagate — the app-level reconciler's re-issued delete is what +// removes them, and does so everywhere. +func TestDeletePrefix_OfflinePeers(t *testing.T) { + ownerKeys, err := accountdata.NewRandom() + require.NoError(t, err) + hubKeys, err := accountdata.NewRandom() + require.NoError(t, err) + hubKeys.SignKey = ownerKeys.SignKey + staleKeys, err := accountdata.NewRandom() + require.NoError(t, err) + staleKeys.SignKey = ownerKeys.SignKey + lateKeys, err := accountdata.NewRandom() + require.NoError(t, err) + lateKeys.SignKey = ownerKeys.SignKey + payload := newStorageCreatePayload(t, ownerKeys) + + fxOwner := newFixture(t, ownerKeys, payload) + fxHub := newFixture(t, hubKeys, payload) + fxStale := newFixture(t, staleKeys, payload) + fxLate := newFixture(t, lateKeys, payload) + hubForOwner := attachPeer(t, fxOwner, ownerKeys, fxHub, hubKeys) + hubForStale := attachPeer(t, fxStale, staleKeys, fxHub, hubKeys) + hubForLate := attachPeer(t, fxLate, lateKeys, fxHub, hubKeys) + + // Stale device writes while offline; owner writes and syncs. + fxStale.add(t, "read/sp1/x", []byte("vx")) + fxStale.add(t, "read/sp1/y", []byte("vy")) + fxStale.add(t, "notes/keep", []byte("vk")) + fxOwner.add(t, "read/sp1/a", []byte("va")) + require.NoError(t, fxOwner.keyValueService.syncWithPeer(ctx, hubForOwner)) + + // Timestamps order the whole scenario; keep them strictly increasing. + time.Sleep(2 * time.Millisecond) + require.NoError(t, fxOwner.defaultStore.DeletePrefix(ctx, "read/sp1/")) + require.NoError(t, fxOwner.keyValueService.syncWithPeer(ctx, hubForOwner)) + require.Len(t, prefixIds(t, fxHub.defaultStore, "read/sp1/"), 1, "hub holds only the watermark") + + // The stale device comes online: one round pushes its rows (rejected) + // and pulls the watermark (drops its local copies). + require.NoError(t, fxStale.keyValueService.syncWithPeer(ctx, hubForStale)) + require.Len(t, prefixIds(t, fxStale.defaultStore, "read/sp1/"), 1, "stale device drops its unsynced rows on receiving the watermark") + require.Len(t, prefixIds(t, fxHub.defaultStore, "read/sp1/"), 1, "stale rows must not stick on the hub") + require.True(t, fxStale.check(t, "notes/keep", []byte("vk")), "unsynced rows outside the prefix survive") + require.True(t, fxHub.check(t, "notes/keep", []byte("vk")), "out-of-prefix rows still sync") + + // A device that writes AFTER the deletion (its clock is past the + // watermark) wins by LWW: the row propagates by design. + time.Sleep(2 * time.Millisecond) + fxLate.add(t, "read/sp1/z", []byte("vz")) + require.NoError(t, fxLate.keyValueService.syncWithPeer(ctx, hubForLate)) + require.Len(t, prefixIds(t, fxHub.defaultStore, "read/sp1/"), 2, "post-deletion write survives the old watermark") + + // The reconciler's re-issued delete removes the late row everywhere. + require.NoError(t, fxOwner.keyValueService.syncWithPeer(ctx, hubForOwner)) + time.Sleep(2 * time.Millisecond) + require.NoError(t, fxOwner.defaultStore.DeletePrefix(ctx, "read/sp1/")) + require.NoError(t, fxOwner.keyValueService.syncWithPeer(ctx, hubForOwner)) + require.NoError(t, fxLate.keyValueService.syncWithPeer(ctx, hubForLate)) + for name, fx := range map[string]*fixture{"owner": fxOwner, "hub": fxHub, "late": fxLate} { + ids := prefixIds(t, fx.defaultStore, "read/sp1/") + require.Len(t, ids, 1, "%s converges to the watermark only: %v", name, ids) + } +}