support async datastores
Adin Schmahmann committed
Dec 5, 2019 at 13:16 UTC
3a0d0e817a8804cd13139465372a7c4f5f764e4c
7 files changed
+132
-8
core/coreapi/unixfs.go
+34
-1
@@ -12,6 +12,7 @@ import (
12
blockservice "github.com/ipfs/go-blockservice"
13
cid "github.com/ipfs/go-cid"
14
cidutil "github.com/ipfs/go-cidutil"
15
+ filestore "github.com/ipfs/go-filestore"
16
bstore "github.com/ipfs/go-ipfs-blockstore"
17
files "github.com/ipfs/go-ipfs-files"
18
ipld "github.com/ipfs/go-ipld-format"
@@ -96,7 +97,29 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
97
bserv := blockservice.New(addblockstore, exch) // hash security 001
98
dserv := dag.NewDAGService(bserv)
99
99
- fileAdder, err := coreunix.NewAdder(ctx, pinning, addblockstore, dserv)
100
+ // add a sync call to the DagService
101
+ // this ensures that data written to the DagService is persisted to the underlying datastore
102
+ // TODO: propagate the Sync function from the datastore through the blockstore, blockservice and dagservice
103
+ var syncDserv *syncDagService
104
+ if settings.OnlyHash {
105
+ syncDserv = &syncDagService{
106
+ DAGService: dserv,
107
+ syncFn: func() error { return nil },
108
+ }
109
+ } else {
110
+ syncDserv = &syncDagService{
111
+ DAGService: dserv,
112
+ syncFn: func() error {
113
+ ds := api.repo.Datastore()
114
+ if err := ds.Sync(bstore.BlockPrefix); err != nil {
115
+ return err
116
+ }
117
+ return ds.Sync(filestore.FilestorePrefix)
118
+ },
119
+ }
120
+ }
121
+
122
+ fileAdder, err := coreunix.NewAdder(ctx, pinning, addblockstore, syncDserv)
123
if err != nil {
124
return nil, err
125
}
@@ -272,3 +295,13 @@ func (api *UnixfsAPI) lsFromLinks(ctx context.Context, ndlinks []*ipld.Link, set
295
func (api *UnixfsAPI) core() *CoreAPI {
296
return (*CoreAPI)(api)
297
}
298
+
299
+// syncDagService is used by the Adder to ensure blocks get persisted to the underlying datastore
300
+type syncDagService struct {
301
+ ipld.DAGService
302
+ syncFn func() error
303
+}
304
+
305
+func (s *syncDagService) Sync() error {
306
+ return s.syncFn()
307
+}
core/coreunix/add.go
+11
@@ -38,6 +38,10 @@ type Link struct {
38
Size uint64
39
}
40
41
+type syncer interface {
42
+ Sync() error
43
+}
44
+
45
// NewAdder Returns a new Adder used for a file add operation.
46
func NewAdder(ctx context.Context, p pin.Pinner, bs bstore.GCLocker, ds ipld.DAGService) (*Adder, error) {
47
bufferedDS := ipld.NewBufferedDAG(ctx, ds)
@@ -316,6 +320,13 @@ func (adder *Adder) AddAllAndPin(file files.Node) (ipld.Node, error) {
320
return nil, err
321
}
322
323
+ if asyncDagService, ok := adder.dagService.(syncer); ok {
324
+ err = asyncDagService.Sync()
325
+ if err != nil {
326
+ return nil, err
327
+ }
328
+ }
329
+
330
if !adder.Pin {
331
return nd, nil
332
}
core/node/core.go
+36
-3
@@ -9,6 +9,7 @@ import (
9
"github.com/ipfs/go-blockservice"
10
"github.com/ipfs/go-cid"
11
"github.com/ipfs/go-datastore"
12
+ "github.com/ipfs/go-filestore"
13
"github.com/ipfs/go-ipfs-blockstore"
14
"github.com/ipfs/go-ipfs-exchange-interface"
15
"github.com/ipfs/go-ipfs-exchange-offline"
@@ -41,18 +42,39 @@ func BlockService(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interf
42
// Pinning creates new pinner which tells GC which blocks should be kept
43
func Pinning(bstore blockstore.Blockstore, ds format.DAGService, repo repo.Repo) (pin.Pinner, error) {
44
internalDag := merkledag.NewDAGService(blockservice.New(bstore, offline.Exchange(bstore)))
44
- pinning, err := pin.LoadPinner(repo.Datastore(), ds, internalDag)
45
+ rootDS := repo.Datastore()
46
+
47
+ syncFn := func() error {
48
+ if err := rootDS.Sync(blockstore.BlockPrefix); err != nil {
49
+ return err
50
+ }
51
+ return rootDS.Sync(filestore.FilestorePrefix)
52
+ }
53
+ syncDs := &syncDagService{ds, syncFn}
54
+ syncInternalDag := &syncDagService{internalDag, syncFn}
55
+
56
+ pinning, err := pin.LoadPinner(rootDS, syncDs, syncInternalDag)
57
if err != nil {
58
// TODO: we should move towards only running 'NewPinner' explicitly on
59
// node init instead of implicitly here as a result of the pinner keys
60
// not being found in the datastore.
61
// this is kinda sketchy and could cause data loss
50
- pinning = pin.NewPinner(repo.Datastore(), ds, internalDag)
62
+ pinning = pin.NewPinner(rootDS, syncDs, syncInternalDag)
63
}
64
65
return pinning, nil
66
}
67
68
+// syncDagService is used by the Pinner to ensure data gets persisted to the underlying datastore
69
+type syncDagService struct {
70
+ format.DAGService
71
+ syncFn func() error
72
+}
73
+
74
+func (s *syncDagService) Sync() error {
75
+ return s.syncFn()
76
+}
77
+
78
// Dag creates new DAGService
79
func Dag(bs blockservice.BlockService) format.DAGService {
80
return merkledag.NewDAGService(bs)
@@ -77,7 +99,18 @@ func OnlineExchange(provide bool) interface{} {
99
func Files(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
100
dsk := datastore.NewKey("/local/filesroot")
101
pf := func(ctx context.Context, c cid.Cid) error {
80
- return repo.Datastore().Put(dsk, c.Bytes())
102
+ rootDS := repo.Datastore()
103
+ if err := rootDS.Sync(blockstore.BlockPrefix); err != nil {
104
+ return err
105
+ }
106
+ if err := rootDS.Sync(filestore.FilestorePrefix); err != nil {
107
+ return err
108
+ }
109
+
110
+ if err := rootDS.Put(dsk, c.Bytes()); err != nil {
111
+ return err
112
+ }
113
+ return rootDS.Sync(dsk)
114
}
115
116
var nd *merkledag.ProtoNode
go.mod
+1
-1
@@ -35,7 +35,7 @@ require (
35
github.com/ipfs/go-ipfs-exchange-interface v0.0.1
36
github.com/ipfs/go-ipfs-exchange-offline v0.0.1
37
github.com/ipfs/go-ipfs-files v0.0.4
38
- github.com/ipfs/go-ipfs-pinner v0.0.2
38
+ github.com/ipfs/go-ipfs-pinner v0.0.3
39
github.com/ipfs/go-ipfs-posinfo v0.0.1
40
github.com/ipfs/go-ipfs-provider v0.3.0
41
github.com/ipfs/go-ipfs-routing v0.1.0
go.sum
+2
-2
@@ -226,8 +226,8 @@ github.com/ipfs/go-ipfs-files v0.0.3/go.mod h1:INEFm0LL2LWXBhNJ2PMIIb2w45hpXgPjN
226
github.com/ipfs/go-ipfs-files v0.0.4 h1:WzRCivcybUQch/Qh6v8LBRhKtRsjnwyiuOV09mK7mrE=
227
github.com/ipfs/go-ipfs-files v0.0.4/go.mod h1:INEFm0LL2LWXBhNJ2PMIIb2w45hpXgPjNoE7yA8Y1d4=
228
github.com/ipfs/go-ipfs-flags v0.0.1/go.mod h1:RnXBb9WV53GSfTrSDVK61NLTFKvWc60n+K9EgCDh+rA=
229
-github.com/ipfs/go-ipfs-pinner v0.0.2 h1:KRXt2V0TzoTd3mO1aONSw8C9wnZtl7RLpPruN/XDnlQ=
230
-github.com/ipfs/go-ipfs-pinner v0.0.2/go.mod h1:KZGyGAR+yLthGEkG9tuA2zweB7O6auXaJNjX6IbEbOs=
229
+github.com/ipfs/go-ipfs-pinner v0.0.3 h1:ez/yNYYyH1W7DiCF/L29tmp6L7lBO8eqbJtPi2pHicA=
230
+github.com/ipfs/go-ipfs-pinner v0.0.3/go.mod h1:s4kFZWLWGDudN8Jyd/GTpt222A12C2snA2+OTdy/7p8=
231
github.com/ipfs/go-ipfs-posinfo v0.0.1 h1:Esoxj+1JgSjX0+ylc0hUmJCOv6V2vFoZiETLR6OtpRs=
232
github.com/ipfs/go-ipfs-posinfo v0.0.1/go.mod h1:SwyeVP+jCwiDu0C313l/8jg6ZxM0qqtlt2a0vILTc1A=
233
github.com/ipfs/go-ipfs-pq v0.0.1 h1:zgUotX8dcAB/w/HidJh1zzc1yFq6Vm8J7T2F4itj/RU=
namesys/publisher.go
+5
-1
@@ -179,7 +179,11 @@ func (p *IpnsPublisher) updateRecord(ctx context.Context, k ci.PrivKey, value pa
179
}
180
181
// Put the new record.
182
- if err := p.ds.Put(IpnsDsKey(id), data); err != nil {
182
+ key := IpnsDsKey(id)
183
+ if err := p.ds.Put(key, data); err != nil {
184
+ return nil, err
185
+ }
186
+ if err := p.ds.Sync(key); err != nil {
187
return nil, err
188
}
189
return entry, nil
namesys/publisher_test.go
+43
@@ -3,6 +3,7 @@ package namesys
3
import (
4
"context"
5
"crypto/rand"
6
+ "github.com/ipfs/go-path"
7
"testing"
8
"time"
9
@@ -110,3 +111,45 @@ func TestRSAPublisher(t *testing.T) {
111
func TestEd22519Publisher(t *testing.T) {
112
testNamekeyPublisher(t, ci.Ed25519, ds.ErrNotFound, false)
113
}
114
+
115
+func TestAsyncDS(t *testing.T) {
116
+ ctx, cancel := context.WithCancel(context.Background())
117
+ defer cancel()
118
+
119
+ rt := mockrouting.NewServer().Client(testutil.RandIdentityOrFatal(t))
120
+ ds := &checkSyncDS{
121
+ Datastore: ds.NewMapDatastore(),
122
+ syncKeys: make(map[ds.Key]struct{}),
123
+ }
124
+ publisher := NewIpnsPublisher(rt, ds)
125
+
126
+ ipnsFakeID := testutil.RandIdentityOrFatal(t)
127
+ ipnsVal, err := path.ParsePath("/ipns/foo.bar")
128
+ if err != nil {
129
+ t.Fatal(err)
130
+ }
131
+
132
+ if err := publisher.Publish(ctx, ipnsFakeID.PrivateKey(), ipnsVal); err != nil {
133
+ t.Fatal(err)
134
+ }
135
+
136
+ ipnsKey := IpnsDsKey(ipnsFakeID.ID())
137
+
138
+ for k := range ds.syncKeys {
139
+ if k.IsAncestorOf(ipnsKey) || k.Equal(ipnsKey) {
140
+ return
141
+ }
142
+ }
143
+
144
+ t.Fatal("ipns key not synced")
145
+}
146
+
147
+type checkSyncDS struct {
148
+ ds.Datastore
149
+ syncKeys map[ds.Key]struct{}
150
+}
151
+
152
+func (d *checkSyncDS) Sync(prefix ds.Key) error {
153
+ d.syncKeys[prefix] = struct{}{}
154
+ return d.Datastore.Sync(prefix)
155
+}