Skip to content

refactor: remove old diffsync (DiffType_V2) support #231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cheggaaa merged 5 commits into from
Aug 7, 2026
Merged

refactor: remove old diffsync (DiffType_V2) support #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6f96158
refactor: remove old diffsync (DiffType_V2) support
cheggaaa Jul 10, 2026
70ca837
fix(nodespace): reject unsupported diff types before deep head sync
cheggaaa Jul 10, 2026
5d4a664
chore: pin any-sync to remove-diffsync-v2 branch (4c9c2983)
cheggaaa Jul 10, 2026
ca54c25
Merge origin/main into remove-diffsync-v2
cheggaaa Aug 7, 2026
9b529dc
chore: drop dead migrationState const, duplicated assertions
cheggaaa Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions archive/adopter/adopter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,10 +152,10 @@ func (ad *adopter) AdoptArchive(ctx context.Context, req *nodesyncproto.AdoptArc
if !ok {
return nil, nodesyncproto.ErrArchiveObjectMissing
}
if err = ad.storage.IndexStorage().MarkArchivedRemote(ctx, req.SpaceId, req.OldHash, req.NewHash, req.CompressedSize, req.UncompressedSize); err != nil {
if err = ad.storage.IndexStorage().MarkArchivedRemote(ctx, req.SpaceId, req.NewHash, req.CompressedSize, req.UncompressedSize); err != nil {
return nil, err
}
if _, err = ad.nodeHead.SetHead(req.SpaceId, req.OldHash, req.NewHash); err != nil {
if _, err = ad.nodeHead.SetHead(req.SpaceId, req.NewHash); err != nil {
log.Warn("can't set nodehead after adoption", zap.String("spaceId", req.SpaceId), zap.Error(err))
err = nil
}
Expand All @@ -173,7 +173,7 @@ func (ad *adopter) AdoptArchive(ctx context.Context, req *nodesyncproto.AdoptArc
// diverged copy could be older than the sender's and must not justify its
// deletion.
func alreadyHaveResponse(entry nodestorage.SpaceStatusEntry, req *nodesyncproto.AdoptArchiveRequest) *nodesyncproto.AdoptArchiveResponse {
if entry.NewHash == req.NewHash && entry.OldHash == req.OldHash {
if entry.NewHash == req.NewHash {
return &nodesyncproto.AdoptArchiveResponse{
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveSame,
}
Expand Down
16 changes: 8 additions & 8 deletions archive/adopter/adopter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,8 @@ func TestAdopter_AdoptArchive(t *testing.T) {
fx.storage.EXPECT().SpaceExists(req.SpaceId).Return(false)
fx.archiveStore.EXPECT().CopyFrom(ctx, req.SrcKey, req.SpaceId).Return(nil)
fx.archiveStore.EXPECT().Exists(ctx, req.SpaceId).Return(true, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "oldHash", "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "oldHash", "newHash").Return(0, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "newHash").Return(0, nil)

resp, err := fx.AdoptArchive(ctx, req)
require.NoError(t, err)
Expand All @@ -76,8 +76,8 @@ func TestAdopter_AdoptArchive(t *testing.T) {
fx.storage.EXPECT().SpaceExists(req.SpaceId).Return(false)
fx.archiveStore.EXPECT().CopyFrom(ctx, req.SrcKey, req.SpaceId).Return(nil)
fx.archiveStore.EXPECT().Exists(ctx, req.SpaceId).Return(true, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "oldHash", "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "oldHash", "newHash").Return(0, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "newHash").Return(0, nil)
fx.archive.EXPECT().QueueRestore(req.SpaceId)

resp, err := fx.AdoptArchive(ctx, req)
Expand All @@ -88,7 +88,7 @@ func TestAdopter_AdoptArchive(t *testing.T) {
fx := newFixture(t)
req := adoptReq()
fx.expectNodePeer()
fx.indexStorage.EXPECT().SpaceStatusEntry(ctx, req.SpaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusOk, OldHash: "oldHash", NewHash: "newHash"}, nil)
fx.indexStorage.EXPECT().SpaceStatusEntry(ctx, req.SpaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusOk, NewHash: "newHash"}, nil)
fx.storage.EXPECT().SpaceExists(req.SpaceId).Return(true)

resp, err := fx.AdoptArchive(ctx, req)
Expand All @@ -99,7 +99,7 @@ func TestAdopter_AdoptArchive(t *testing.T) {
fx := newFixture(t)
req := adoptReq()
fx.expectNodePeer()
fx.indexStorage.EXPECT().SpaceStatusEntry(ctx, req.SpaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusArchived, OldHash: "oldHash", NewHash: "newHash"}, nil)
fx.indexStorage.EXPECT().SpaceStatusEntry(ctx, req.SpaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusArchived, NewHash: "newHash"}, nil)
fx.archiveStore.EXPECT().Exists(ctx, req.SpaceId).Return(true, nil)

resp, err := fx.AdoptArchive(ctx, req)
Expand All @@ -114,8 +114,8 @@ func TestAdopter_AdoptArchive(t *testing.T) {
fx.archiveStore.EXPECT().Exists(ctx, req.SpaceId).Return(false, nil)
fx.archiveStore.EXPECT().CopyFrom(ctx, req.SrcKey, req.SpaceId).Return(nil)
fx.archiveStore.EXPECT().Exists(ctx, req.SpaceId).Return(true, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "oldHash", "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "oldHash", "newHash").Return(0, nil)
fx.indexStorage.EXPECT().MarkArchivedRemote(ctx, req.SpaceId, "newHash", int64(10), int64(20)).Return(nil)
fx.nodeHead.EXPECT().SetHead(req.SpaceId, "newHash").Return(0, nil)

resp, err := fx.AdoptArchive(ctx, req)
require.NoError(t, err)
Expand Down
26 changes: 12 additions & 14 deletions archive/archive.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,12 +47,12 @@ type Archive interface {
// keeping the local DB and its status intact. Used to ship live spaces
// through the shared archive store during migration. The caller must ensure
// the space is not already archived: opening an archived space triggers a
// restore. Returns the heads read from the snapshot itself (they exactly
// describe the uploaded object, unlike the asynchronously updated index)
// restore. Returns the heads hash read from the snapshot itself (it exactly
// describes the uploaded object, unlike the asynchronously updated index)
// and the snapshot sizes.
// The return types are primitives on purpose: a struct would make the
// generated mock import this package and create test-only import cycles.
ForceArchive(ctx context.Context, spaceId string) (oldHash, newHash string, compressedSize, uncompressedSize int64, err error)
ForceArchive(ctx context.Context, spaceId string) (hash string, compressedSize, uncompressedSize int64, err error)
// QueueRestore schedules a background restore of an archived space
// (used for eager adoption of migrated spaces).
QueueRestore(spaceId string)
Expand Down Expand Up @@ -213,12 +213,12 @@ func (a *archive) Archive(ctx context.Context, spaceId string) (err error) {
return
}

func (a *archive) ForceArchive(ctx context.Context, spaceId string) (oldHash, newHash string, compressedSize, uncompressedSize int64, err error) {
func (a *archive) ForceArchive(ctx context.Context, spaceId string) (hash string, compressedSize, uncompressedSize int64, err error) {
// DumpStorage backups the db into a temp dir; it works for open spaces too
err = a.storageProvider.DumpStorage(ctx, spaceId, func(path string) error {
// read the heads from the snapshot: they describe exactly what the
// uploaded object contains
oldHash, newHash, err = readSnapshotHeads(ctx, spaceId, filepath.Join(path, "store.db"))
hash, err = readSnapshotHead(ctx, spaceId, filepath.Join(path, "store.db"))
if err != nil {
return err
}
Expand All @@ -243,10 +243,10 @@ func (a *archive) ForceArchive(ctx context.Context, spaceId string) (oldHash, ne
return
}

// readSnapshotHeads reads the space state directly from the snapshot db
// readSnapshotHead reads the space state directly from the snapshot db
// (schema of commonspace/headsync/statestorage, including the legacy
// single-hash fallback).
func readSnapshotHeads(ctx context.Context, spaceId, dbPath string) (oldHash, newHash string, err error) {
// "h" key fallback).
func readSnapshotHead(ctx context.Context, spaceId, dbPath string) (hash string, err error) {
db, err := anystore.Open(ctx, dbPath, nil)
if err != nil {
return
Expand All @@ -262,13 +262,11 @@ func readSnapshotHeads(ctx context.Context, spaceId, dbPath string) (oldHash, ne
if err != nil {
return
}
oldHash = doc.Value().GetString("oh")
newHash = doc.Value().GetString("nh")
if oldHash == "" || newHash == "" {
oldHash = doc.Value().GetString("h")
newHash = oldHash
hash = doc.Value().GetString("nh")
if hash == "" {
hash = doc.Value().GetString("h")
}
return oldHash, newHash, nil
return hash, nil
}

// createGzipFromStore creates store.gz from store.db inside spaceDir.
Expand Down
5 changes: 2 additions & 3 deletions archive/archive_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,10 +173,9 @@ func TestArchive_ForceArchive(t *testing.T) {
})

// no MarkArchived, no local deletion: the space stays live
oldHash, newHash, compressedSize, uncompressedSize, err := fx.ForceArchive(ctx, spaceId)
hash, compressedSize, uncompressedSize, err := fx.ForceArchive(ctx, spaceId)
require.NoError(t, err)
assert.Equal(t, "snap-old", oldHash)
assert.Equal(t, "snap-new", newHash)
assert.Equal(t, "snap-new", hash)
assert.Greater(t, compressedSize, int64(0))
assert.Greater(t, uncompressedSize, int64(0))
assert.Equal(t, int(compressedSize), uploaded)
Expand Down
9 changes: 4 additions & 5 deletions archive/mock_archive/mock_archive.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

23 changes: 4 additions & 19 deletions nodehead/mock_nodehead/mock_nodehead.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

27 changes: 6 additions & 21 deletions nodehead/nodehead.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,8 @@ func New() NodeHead {

// NodeHead keeps current state of all spaces by partitions
type NodeHead interface {
SetHead(spaceId, oldHead, newHead string) (part int, err error)
SetHead(spaceId, head string) (part int, err error)
GetHead(spaceId string) (head string, err error)
GetOldHead(spaceId string) (head string, err error)
DeleteHeads(spaceId string) error
ReloadHeadFromStore(ctx context.Context, spaceId string) error
LDiff(partId int) ldiff.Diff
Expand All @@ -53,18 +52,16 @@ type nodeStorage interface {
type nodeHead struct {
mu sync.Mutex
partitions map[int]ldiff.Diff
oldHashes map[string]string
nodeconf nodeconf.NodeConf
spaceStore nodeStorage
}

func (n *nodeHead) Init(a *app.App) (err error) {
n.partitions = map[int]ldiff.Diff{}
n.oldHashes = map[string]string{}
n.nodeconf = a.MustComponent(nodeconf.CName).(nodeconf.NodeConf)
n.spaceStore = a.MustComponent(spacestorage.CName).(nodeStorage)
n.spaceStore.OnWriteHash(func(_ context.Context, spaceId, oldHash, newHash string) {
if _, e := n.SetHead(spaceId, oldHash, newHash); e != nil {
n.spaceStore.OnWriteHash(func(_ context.Context, spaceId, hash string) {
if _, e := n.SetHead(spaceId, hash); e != nil {
log.Error("can't set head", zap.Error(e))
}
})
Expand All @@ -88,7 +85,7 @@ func (n *nodeHead) Run(ctx context.Context) (err error) {
var total int
err = n.spaceStore.IndexStorage().ReadHashes(ctx, func(update nodestorage.SpaceUpdate) (bool, error) {
total++
if _, e := n.SetHead(update.SpaceId, update.OldHash, update.NewHash); e != nil {
if _, e := n.SetHead(update.SpaceId, update.NewHash); e != nil {
log.Error("can't set head", zap.Error(e))
return false, e
}
Expand All @@ -104,15 +101,14 @@ func (n *nodeHead) Run(ctx context.Context) (err error) {
func (n *nodeHead) DeleteHeads(spaceId string) error {
n.mu.Lock()
defer n.mu.Unlock()
delete(n.oldHashes, spaceId)
part := n.nodeconf.Partition(spaceId)
if ld, ok := n.partitions[part]; ok {
return ld.RemoveId(spaceId)
}
return nil
}

func (n *nodeHead) SetHead(spaceId, oldHead, newHead string) (part int, err error) {
func (n *nodeHead) SetHead(spaceId, head string) (part int, err error) {
part = n.nodeconf.Partition(spaceId)
n.mu.Lock()
defer n.mu.Unlock()
Expand All @@ -121,8 +117,7 @@ func (n *nodeHead) SetHead(spaceId, oldHead, newHead string) (part int, err erro
ld = ldiff.New(16, 16)
n.partitions[part] = ld
}
ld.Set(ldiff.Element{Id: spaceId, Head: newHead})
n.oldHashes[spaceId] = oldHead
ld.Set(ldiff.Element{Id: spaceId, Head: head})
return
}

Expand Down Expand Up @@ -160,16 +155,6 @@ func (n *nodeHead) GetHead(spaceId string) (hash string, err error) {
return el.Head, nil
}

func (n *nodeHead) GetOldHead(spaceId string) (hash string, err error) {
n.mu.Lock()
defer n.mu.Unlock()
hash, ok := n.oldHashes[spaceId]
if !ok {
err = ErrSpaceNotFound
}
return
}

func (n *nodeHead) ReloadHeadFromStore(ctx context.Context, spaceId string) error {
ss, err := n.spaceStore.IndexSpace(ctx, spaceId, true)
if err != nil {
Expand Down
14 changes: 4 additions & 10 deletions nodehead/nodehead_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,6 @@ func TestNodeHead_Run(t *testing.T) {
fx = newFixture(t, tmpDir)
defer fx.Finish(t)
assert.Len(t, fx.NodeHead.(*nodeHead).partitions, 1)
assert.Len(t, fx.NodeHead.(*nodeHead).oldHashes, 10)
})
}

Expand All @@ -80,10 +79,10 @@ func TestNodeHead_SetHead(t *testing.T) {
}

t.Run("set head", func(t *testing.T) {
part, err := fx.SetHead("2.2", "oldHead", "newhead")
part, err := fx.SetHead("2.2", "newhead")
require.NoError(t, err)
h1 := getHash(part)
part2, err := fx.SetHead("3.2", "oldHead", "newhead")
part2, err := fx.SetHead("3.2", "newhead")
assert.Equal(t, part, part2)
h2 := getHash(part)
assert.NotEqual(t, h1, h2)
Expand All @@ -109,19 +108,14 @@ func TestNodeHead_Ranges(t *testing.T) {
func TestNodeHead_GetSpaceHash(t *testing.T) {
fx := newFixture(t, "")
defer fx.Finish(t)
oldHash := "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262"
newHash := "bf1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262"
_, err := fx.SetHead("space1", oldHash, newHash)
_, err := fx.SetHead("space1", newHash)
require.NoError(t, err)

head, err := fx.GetHead("space1")
require.NoError(t, err)
assert.Equal(t, newHash, head)

head, err = fx.GetOldHead("space1")
require.NoError(t, err)
assert.Equal(t, oldHash, head)

_, err = fx.GetHead("not found")
assert.Equal(t, ErrSpaceNotFound, err)
}
Expand All @@ -130,7 +124,7 @@ func TestNodeHead_DeleteHeads(t *testing.T) {
fx := newFixture(t, "")
defer fx.Finish(t)
hash := "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262"
_, err := fx.SetHead("space1", hash, hash)
_, err := fx.SetHead("space1", hash)
require.NoError(t, err)

err = fx.NodeHead.(*nodeHead).DeleteHeads("space1")
Expand Down
3 changes: 0 additions & 3 deletions nodespace/migrator/migrator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,9 +130,6 @@ func (m *migrator) Run(ctx context.Context) (err error) {
}
err = m.newStorage.IndexStorage().UpdateHash(ctx, nodestorage.SpaceUpdate{
SpaceId: id,
// statestorage.State no longer carries a distinct old hash after the
// diffsync-V2 removal; use NewHash for both (matches the "oh" == "nh" mirror).
OldHash: state.NewHash,
NewHash: state.NewHash,
})
if err != nil {
Expand Down
Loading
Loading