Skip to content

Commit ecb221d

Browse files
authored
Merge pull request #238 from anyproto/cheggaaa/syn-131-any-sync-0.13.1
chore: any-sync v0.13.1
2 parents 97d7805 + da971d5 commit ecb221d

54 files changed

Lines changed: 5801 additions & 246 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎README.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,13 @@ The following options are available for running the Any-Sync Node:
2424
- `-v` — current version.
2525
- `-h` — help message.
2626
27+
## Changing the node set (resharding)
28+
29+
Adding and removing tree nodes on a running network is supported when the
30+
nodes share an archive bucket — see the operator guide:
31+
[docs/resharding-operations.md](docs/resharding-operations.md)
32+
(design: [docs/resharding-plan.md](docs/resharding-plan.md)).
33+
2734
## Graph example of using Any-Sync Nodes group
2835
2936
```mermaid

‎archive/adopter/adopter.go‎

Lines changed: 226 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,226 @@
1+
//go:generate mockgen -destination mock_adopter/mock_adopter.go github.com/anyproto/any-sync-node/archive/adopter Adopter
2+
package adopter
3+
4+
import (
5+
"context"
6+
"errors"
7+
8+
anystore "github.com/anyproto/any-store"
9+
"github.com/anyproto/any-sync/app"
10+
"github.com/anyproto/any-sync/app/logger"
11+
"github.com/anyproto/any-sync/metric"
12+
"github.com/anyproto/any-sync/net/peer"
13+
"github.com/anyproto/any-sync/nodeconf"
14+
"go.uber.org/zap"
15+
16+
"github.com/anyproto/any-sync-node/archive"
17+
"github.com/anyproto/any-sync-node/archive/archivestore"
18+
"github.com/anyproto/any-sync-node/nodehead"
19+
"github.com/anyproto/any-sync-node/nodestorage"
20+
"github.com/anyproto/any-sync-node/nodesync/nodesyncproto"
21+
)
22+
23+
const CName = "node.archive.adopter"
24+
25+
var log = logger.NewNamed(CName)
26+
27+
func New() Adopter {
28+
return new(adopter)
29+
}
30+
31+
// Adopter handles AdoptArchive requests: another node hands over an archived
32+
// space snapshot parked in the shared bucket; we server-side copy the object
33+
// into our own prefix and register the space as archived without ever
34+
// materializing the SQLite DB.
35+
type Adopter interface {
36+
app.Component
37+
AdoptArchive(ctx context.Context, req *nodesyncproto.AdoptArchiveRequest) (resp *nodesyncproto.AdoptArchiveResponse, err error)
38+
}
39+
40+
// historyPeerEpochs is how many retained configuration epochs back a sender
41+
// may be recognized as a network node: a node removed from the configuration
42+
// must still be able to hand its spaces off while it drains.
43+
const historyPeerEpochs = 5
44+
45+
type adopter struct {
46+
storage nodestorage.NodeStorage
47+
archiveStore archivestore.ArchiveStore
48+
archive archive.Archive
49+
nodeHead nodehead.NodeHead
50+
nodeConf nodeconf.Service
51+
confHistory nodeconf.HistoryStore
52+
stat *adopterStat
53+
}
54+
55+
func (ad *adopter) Init(a *app.App) (err error) {
56+
ad.storage = a.MustComponent(nodestorage.CName).(nodestorage.NodeStorage)
57+
ad.archiveStore = a.MustComponent(archivestore.CName).(archivestore.ArchiveStore)
58+
ad.archive = a.MustComponent(archive.CName).(archive.Archive)
59+
ad.nodeHead = a.MustComponent(nodehead.CName).(nodehead.NodeHead)
60+
ad.nodeConf = a.MustComponent(nodeconf.CName).(nodeconf.Service)
61+
ad.confHistory, _ = a.Component(nodeconf.CNameStore).(nodeconf.HistoryStore)
62+
ad.stat = new(adopterStat)
63+
if m := a.Component(metric.CName); m != nil {
64+
registerMetric(ad.stat, m.(metric.Metric).Registry())
65+
}
66+
return
67+
}
68+
69+
func (ad *adopter) Name() (name string) {
70+
return CName
71+
}
72+
73+
func (ad *adopter) AdoptArchive(ctx context.Context, req *nodesyncproto.AdoptArchiveRequest) (resp *nodesyncproto.AdoptArchiveResponse, err error) {
74+
defer func() {
75+
switch {
76+
case err != nil:
77+
ad.stat.rejected.Add(1)
78+
case resp.Result == nodesyncproto.AdoptArchiveResult_AdoptArchiveOk:
79+
ad.stat.adopted.Add(1)
80+
case resp.Result == nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveSame:
81+
ad.stat.alreadyHaveSame.Add(1)
82+
case resp.Result == nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveDiverged:
83+
ad.stat.alreadyHaveDiverged.Add(1)
84+
}
85+
}()
86+
if !ad.archiveStore.Shared() {
87+
return nil, nodesyncproto.ErrArchiveUnavailable
88+
}
89+
if err = ad.checkPeerIsNode(ctx); err != nil {
90+
return nil, err
91+
}
92+
if !ad.nodeConf.IsResponsible(req.SpaceId) {
93+
// the sender acts on a different (stale or newer) configuration:
94+
// never adopt spaces we don't own — an ACK from a non-owner must not
95+
// justify the sender's deletion
96+
return nil, nodesyncproto.ErrNotResponsible
97+
}
98+
entry, entryErr := ad.storage.IndexStorage().SpaceStatusEntry(ctx, req.SpaceId)
99+
switch {
100+
case entryErr == nil:
101+
switch entry.Status {
102+
case nodestorage.SpaceStatusRemove:
103+
return nil, nodesyncproto.ErrSpaceDeleted
104+
case nodestorage.SpaceStatusRemovePrepare:
105+
// pending deletion is cancellable: don't let the sender drop its
106+
// copy, and don't adopt data into a deletion-flow status either;
107+
// the sender parks the space until the deletion resolves
108+
return nil, nodesyncproto.ErrSpacePendingDeletion
109+
case nodestorage.SpaceStatusOk:
110+
if ad.storage.SpaceExists(req.SpaceId) {
111+
return alreadyHaveResponse(entry, req), nil
112+
}
113+
// index entry says ok, but there is no local db: adopt to repair
114+
case nodestorage.SpaceStatusArchived:
115+
ok, hErr := ad.archiveStore.Exists(ctx, req.SpaceId)
116+
if hErr != nil {
117+
return nil, hErr
118+
}
119+
if ok {
120+
return alreadyHaveResponse(entry, req), nil
121+
}
122+
// index says archived, but our object is gone: adopt to repair
123+
case nodestorage.SpaceStatusError:
124+
// an Error entry needs an operator: the local db may hold data
125+
// that failed to archive — never overwrite it with a snapshot and
126+
// never ACK the sender's deletion with a broken copy
127+
return &nodesyncproto.AdoptArchiveResponse{
128+
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveDiverged,
129+
}, nil
130+
}
131+
case errors.Is(entryErr, anystore.ErrDocNotFound):
132+
if ad.storage.SpaceExists(req.SpaceId) {
133+
// a local db without an index entry: heads unknown, force convergence
134+
return &nodesyncproto.AdoptArchiveResponse{
135+
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveDiverged,
136+
}, nil
137+
}
138+
default:
139+
return nil, entryErr
140+
}
141+
142+
if err = ad.archiveStore.CopyFrom(ctx, req.SrcKey, req.SpaceId); err != nil {
143+
if errors.Is(err, archivestore.ErrNotFound) {
144+
return nil, nodesyncproto.ErrArchiveObjectMissing
145+
}
146+
return nil, err
147+
}
148+
ok, err := ad.archiveStore.Exists(ctx, req.SpaceId)
149+
if err != nil {
150+
return nil, err
151+
}
152+
if !ok {
153+
return nil, nodesyncproto.ErrArchiveObjectMissing
154+
}
155+
if err = ad.storage.IndexStorage().MarkArchivedRemote(ctx, req.SpaceId, req.OldHash, req.NewHash, req.CompressedSize, req.UncompressedSize); err != nil {
156+
return nil, err
157+
}
158+
if _, err = ad.nodeHead.SetHead(req.SpaceId, req.OldHash, req.NewHash); err != nil {
159+
log.Warn("can't set nodehead after adoption", zap.String("spaceId", req.SpaceId), zap.Error(err))
160+
err = nil
161+
}
162+
log.Info("adopted archived space", zap.String("spaceId", req.SpaceId), zap.Bool("eager", req.Eager))
163+
if req.Eager {
164+
ad.archive.QueueRestore(req.SpaceId)
165+
}
166+
return &nodesyncproto.AdoptArchiveResponse{
167+
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveOk,
168+
}, nil
169+
}
170+
171+
// alreadyHaveResponse reports whether our existing copy matches the offered
172+
// heads: only an exact match counts as a durable ACK for the sender — a
173+
// diverged copy could be older than the sender's and must not justify its
174+
// deletion.
175+
func alreadyHaveResponse(entry nodestorage.SpaceStatusEntry, req *nodesyncproto.AdoptArchiveRequest) *nodesyncproto.AdoptArchiveResponse {
176+
if entry.NewHash == req.NewHash && entry.OldHash == req.OldHash {
177+
return &nodesyncproto.AdoptArchiveResponse{
178+
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveSame,
179+
}
180+
}
181+
return &nodesyncproto.AdoptArchiveResponse{
182+
Result: nodesyncproto.AdoptArchiveResult_AdoptArchiveAlreadyHaveDiverged,
183+
}
184+
}
185+
186+
func (ad *adopter) checkPeerIsNode(ctx context.Context) (err error) {
187+
peerId, err := peer.CtxPeerId(ctx)
188+
if err != nil {
189+
return err
190+
}
191+
if len(ad.nodeConf.NodeTypes(peerId)) > 0 {
192+
return nil
193+
}
194+
// a node removed from the configuration keeps draining after the removal:
195+
// recognize senders that were tree nodes in recently retained epochs
196+
if ad.wasRecentTreeNode(ctx, peerId) {
197+
return nil
198+
}
199+
return nodesyncproto.ErrPeerIsNotNode
200+
}
201+
202+
func (ad *adopter) wasRecentTreeNode(ctx context.Context, peerId string) bool {
203+
if ad.confHistory == nil {
204+
return false
205+
}
206+
netId := ad.nodeConf.Configuration().NetworkId
207+
epochs, err := ad.confHistory.Epochs(ctx, netId)
208+
if err != nil || len(epochs) == 0 {
209+
return false
210+
}
211+
if len(epochs) > historyPeerEpochs {
212+
epochs = epochs[len(epochs)-historyPeerEpochs:]
213+
}
214+
for i := len(epochs) - 1; i >= 0; i-- {
215+
conf, gErr := ad.confHistory.GetByEpoch(ctx, netId, epochs[i])
216+
if gErr != nil {
217+
continue
218+
}
219+
for _, n := range conf.Nodes {
220+
if n.PeerId == peerId && n.HasType(nodeconf.NodeTypeTree) {
221+
return true
222+
}
223+
}
224+
}
225+
return false
226+
}

0 commit comments

Comments
 (0)