Merge pull request #114 from berty/d4ryl00/fix/lifecycle

fix: close of stores

Author: D4ryl00

.github/workflows/go.yml

@@ -16,19 +16,24 @@ jobs:
         golang:
           - 1.18.x
     steps:
-      - uses: actions/checkout@v3
-      - name: Set up Go
-        uses: actions/setup-go@v2
-        with:
-          go-version: ${{ matrix.golang }}
-      - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3.3.0
-        with:
-          go-version: ${{ matrix.golang }}
-          version: v1.50.1
-          args: --timeout=10m
-          # only-new-issues: true
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Setup asdf
+        uses: asdf-vm/actions/setup@v1
+
+      - name: Setup golang
+        run: |
+          asdf plugin add golang
+          asdf install golang
+
+      - name: Setup golangci-lint
+        run: |
+          asdf plugin add golangci-lint
+          asdf install golangci-lint
 
+      - name: Run golangci-lint
+        run: make lint
 
   go-tests-on-linux:
     runs-on: ubuntu-latest

.tool-versions

@@ -1 +1,2 @@
 golang 1.18
+golangci-lint 1.50.1

baseorbitdb/events.go

@@ -346,6 +346,11 @@ type NewStoreOptions struct {
 	Tracer                 trace.Tracer
 	IO                     ipfslog.IO
 	StoreSpecificOpts      interface{}
+	PubSub                 PubSubInterface
+	MessageMarshaler       MessageMarshaler
+	PeerID                 peer.ID
+	DirectChannelFactory   DirectChannelFactory
+	NewHeadsEmitter        event.Emitter
 }
 
 type DirectChannelOptions struct {
@@ -371,7 +376,7 @@ type DirectChannelEmitter interface {
 type DirectChannelFactory func(ctx context.Context, emitter DirectChannelEmitter, opts *DirectChannelOptions) (DirectChannel, error)
 
 // StoreConstructor Defines the expected constructor for a custom store
-type StoreConstructor func(context.Context, coreapi.CoreAPI, *identityprovider.Identity, address.Address, *NewStoreOptions) (Store, error)
+type StoreConstructor func(coreapi.CoreAPI, *identityprovider.Identity, address.Address, *NewStoreOptions) (Store, error)
 
 // IndexConstructor Defines the expected constructor for a custom index
 type IndexConstructor func(publicKey []byte) StoreIndex

baseorbitdb/events_handler.go

@@ -0,0 +1,2 @@
+// Package messagemarshaler is a package for marshal messages
+package messagemarshaler // import "berty.tech/go-orbit-db/messagemarshaler"

baseorbitdb/orbitdb.go

@@ -0,0 +1,17 @@
+package messagemarshaler
+
+import (
+	"encoding/json"
+
+	"berty.tech/go-orbit-db/iface"
+)
+
+type JSONMarshaler struct{}
+
+func (JSONMarshaler) Marshal(m *iface.MessageExchangeHeads) ([]byte, error) {
+	return json.Marshal(m)
+}
+
+func (JSONMarshaler) Unmarshal(data []byte, m *iface.MessageExchangeHeads) error {
+	return json.Unmarshal(data, m)
+}

iface/interface.go

@@ -23,8 +23,10 @@ const (
 )
 
 type channel struct {
-	id  string
-	sub coreapi.PubSubSubscription
+	ctx    context.Context
+	cancel context.CancelFunc
+	id     string
+	sub    coreapi.PubSubSubscription
 }
 
 type channels struct {
@@ -52,12 +54,16 @@ func (c *channels) Connect(ctx context.Context, target peer.ID) error {
 			return fmt.Errorf("unable to subscribe to pubsub: %w", err)
 		}
 
+		ctx, cancel := context.WithCancel(ctx)
+
 		c.subs[target] = &channel{
-			sub: sub,
-			id:  id,
+			ctx:    ctx,
+			cancel: cancel,
+			sub:    sub,
+			id:     id,
 		}
 		go func() {
-			c.monitorTopic(sub, target)
+			c.monitorTopic(ctx, sub, target)
 
 			// if monitor topic is done, remove target from cache
 			c.muSubs.Lock()
@@ -127,9 +133,9 @@ func (c *channels) getChannelID(p peer.ID) string {
 	return fmt.Sprintf("/%s/%s", PROTOCOL, strings.Join(channelIDPeers, "/"))
 }
 
-func (c *channels) monitorTopic(sub coreapi.PubSubSubscription, p peer.ID) {
+func (c *channels) monitorTopic(ctx context.Context, sub coreapi.PubSubSubscription, p peer.ID) {
 	for {
-		msg, err := sub.Next(c.ctx)
+		msg, err := sub.Next(ctx)
 		switch err {
 		case nil:
 		case context.Canceled, context.DeadlineExceeded:

messagemarshaler/doc.go

@@ -36,6 +36,7 @@ func (p *psTopic) WatchPeers(ctx context.Context) (<-chan events.Event, error) {
 
 	ch := make(chan events.Event, 32)
 	go func() {
+		defer ph.Cancel()
 		defer close(ch)
 		for {
 			evt, err := ph.NextPeerEvent(ctx)
@@ -70,6 +71,7 @@ func (p *psTopic) WatchMessages(ctx context.Context) (<-chan *iface.EventPubSubM
 	ch := make(chan *iface.EventPubSubMessage, 32)
 	go func() {
 		defer close(ch)
+		defer sub.Cancel()
 		for {
 			msg, err := sub.Next(ctx)
 			if err != nil {

messagemarshaler/json_marshaler.go

@@ -242,7 +242,7 @@ func DefaultStoreOptsForMap(keyField string) *iface.CreateDocumentDBOptions {
 }
 
 // NewOrbitDBDocumentStore Instantiates a new DocumentStore
-func NewOrbitDBDocumentStore(ctx context.Context, ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (iface.Store, error) {
+func NewOrbitDBDocumentStore(ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (iface.Store, error) {
 	if options.StoreSpecificOpts == nil {
 		options.StoreSpecificOpts = DefaultStoreOptsForMap("_id")
 	}
@@ -271,7 +271,7 @@ func NewOrbitDBDocumentStore(ctx context.Context, ipfs coreapi.CoreAPI, identity
 	store := &orbitDBDocumentStore{docOpts: docOpts}
 	options.Index = func(_ []byte) iface.StoreIndex { return newDocumentIndex(docOpts) }
 
-	err := store.InitBaseStore(ctx, ipfs, identity, addr, options)
+	err := store.InitBaseStore(ipfs, identity, addr, options)
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize document store: %w", err)
 	}

pubsub/oneonone/channel.go

@@ -201,11 +201,12 @@ func (o *orbitDBEventLogStore) Type() string {
 }
 
 // NewOrbitDBEventLogStore Instantiates a new EventLogStore
-func NewOrbitDBEventLogStore(ctx context.Context, ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (i iface.Store, e error) {
+func NewOrbitDBEventLogStore(ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (i iface.Store, e error) {
+
 	store := &orbitDBEventLogStore{}
 	options.Index = NewEventIndex
 
-	err := store.InitBaseStore(ctx, ipfs, identity, addr, options)
+	err := store.InitBaseStore(ipfs, identity, addr, options)
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize base store: %w", err)
 	}

pubsub/pubsubraw/pubsub.go

@@ -3,6 +3,7 @@ package stores
 import (
 	ipfslog "berty.tech/go-ipfs-log"
 	"berty.tech/go-orbit-db/address"
+	"berty.tech/go-orbit-db/iface"
 	"berty.tech/go-orbit-db/stores/replicator"
 	cid "github.com/ipfs/go-cid"
 	peer "github.com/libp2p/go-libp2p/core/peer"
@@ -15,6 +16,7 @@ var Events = []interface{}{
 	new(EventLoad),
 	new(EventReplicated),
 	new(EventReplicate),
+	new(EventExchangeHeads),
 }
 
 type EventReplicate struct {
@@ -137,3 +139,16 @@ func NewEventNewPeer(p peer.ID) EventNewPeer {
 		Peer: p,
 	}
 }
+
+// EventExchangeHeadsset An event as stateful, sent when new exchange head is done, so newly subscriber can replay last event in case they missed it
+type EventExchangeHeads struct {
+	Peer    peer.ID
+	Message *iface.MessageExchangeHeads
+}
+
+func NewEventExchangeHeads(p peer.ID, msg *iface.MessageExchangeHeads) EventExchangeHeads {
+	return EventExchangeHeads{
+		Peer:    p,
+		Message: msg,
+	}
+}

stores/basestore/base_store.go

@@ -85,12 +85,12 @@ func (o *orbitDBKeyValue) Type() string {
 }
 
 // NewOrbitDBKeyValue Instantiates a new KeyValueStore
-func NewOrbitDBKeyValue(ctx context.Context, ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (i iface.Store, e error) {
+func NewOrbitDBKeyValue(ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) (i iface.Store, e error) {
 	store := &orbitDBKeyValue{}
 
 	options.Index = NewKVIndex
 
-	err := store.InitBaseStore(ctx, ipfs, identity, addr, options)
+	err := store.InitBaseStore(ipfs, identity, addr, options)
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize base store: %w", err)
 	}

stores/documentstore/document.go

@@ -384,11 +384,11 @@ func testDirectChannelNodeGenerator(t *testing.T, mn mocknet.Mocknet, i int) (or
 
 	orbitdb1, err := orbitdb.NewOrbitDB(ctx, ipfs1, &orbitdb.NewOrbitDBOptions{
 		Directory:            &dbPath1,
+		Logger:               logger,
 		DirectChannelFactory: directchannel.InitDirectChannelFactory(zap.NewNop(), node1.PeerHost),
 		// @NOTE(gfanton): use raw pubsub here, we need a higher buffer
 		// for subscribe to make the test works on CI
 		PubSub: pubsubraw.NewPubSub(node1.PubSub, node1.Identity, nil, nil),
-		Logger: logger,
 	})
 	require.NoError(t, err)
 
@@ -422,11 +422,11 @@ func testDefaultNodeGenerator(t *testing.T, mn mocknet.Mocknet, i int) (orbitdb.
 	logger.Named("orbitdb.tests").Debug(fmt.Sprintf("node%d is %s", i, node1.Identity.String()))
 
 	orbitdb1, err := orbitdb.NewOrbitDB(ctx, ipfs1, &orbitdb.NewOrbitDBOptions{
-		// @NOTE(gfanton): use raw pubsub here, we need a higher buffer
-		// for subscribe to make the test works on CI
-		PubSub:    pubsubraw.NewPubSub(node1.PubSub, node1.Identity, logger, nil),
 		Directory: &dbPath1,
 		Logger:    logger,
+		// @NOTE(gfanton): use raw pubsub here, we need a higher buffer
+		// for subscribe to make the test works on CI
+		PubSub: pubsubraw.NewPubSub(node1.PubSub, node1.Identity, nil, nil),
 	})
 	require.NoError(t, err)