feat: BREAKING CHANGE: orbitdb.Open and StoreConstructor

D4ryl00 · D4ryl00 · commit 7a745623bd51 · 2022-11-17T16:10:36.000+01:00
Signed-off-by: D4ryl00 &lt;d4ryl00@gmail.com&gt;
diff --git a/baseorbitdb/orbitdb.go b/baseorbitdb/orbitdb.go
@@ -98,6 +98,8 @@ type NewOrbitDBOptions struct {
 }
 
 type orbitDB struct {
+	ctx                   context.Context
+	cancel                context.CancelFunc
 	storeTypes            map[string]iface.StoreConstructor
 	accessControllerTypes map[string]iface.AccessControllerConstructor
 	ipfs                  coreapi.CoreAPI
@@ -197,6 +199,12 @@ func (o *orbitDB) getStore(address string) (iface.Store, bool) {
 
 	store, ok := o.stores[address]
 
+	if ok && store.IsClosed() {
+		o.logger.Debug("getStore: store is closed and will be deleted")
+		delete(o.stores, address)
+		return nil, false
+	}
+
 	return store, ok
 }
 
@@ -376,7 +384,10 @@ func newOrbitDB(ctx context.Context, is coreapi.CoreAPI, identity *idp.Identity,
 		options.Directory = &cacheleveldown.InMemoryDirectory
 	}
 
+	ctx, cancel := context.WithCancel(ctx)
 	odb := &orbitDB{
+		ctx:                   ctx,
+		cancel:                cancel,
 		ipfs:                  is,
 		identity:              identity,
 		id:                    *options.PeerID,
@@ -397,10 +408,12 @@ func newOrbitDB(ctx context.Context, is coreapi.CoreAPI, identity *idp.Identity,
 	// set new heads as stateful, so newly subscriber can replay last event in case they missed it
 	odb.emitters.newHeads, err = eventBus.Emitter(new(EventExchangeHeads), eventbus.Stateful)
 	if err != nil {
+		cancel()
 		return nil, fmt.Errorf("unable to create global emitter: %w", err)
 	}
 
 	if err := odb.monitorDirectChannel(ctx, eventBus); err != nil {
+		cancel()
 		return nil, fmt.Errorf("unable to monitor direct channel: %w", err)
 	}
 
@@ -480,6 +493,7 @@ func (o *orbitDB) Close() error {
 	o.closeCache()
 	o.closeKeyStore()
 
+	o.cancel()
 	return nil
 }
 
@@ -755,7 +769,8 @@ func (o *orbitDB) createStore(ctx context.Context, storeType string, parsedDBAdd
 		options.Directory = &o.directory
 	}
 
-	store, err := storeFunc(ctx, o.IPFS(), identity, parsedDBAddress, &iface.NewStoreOptions{
+	store, err := storeFunc(o.IPFS(), identity, parsedDBAddress, &iface.NewStoreOptions{
+		EventBus:          options.EventBus,
 		AccessController:  accessController,
 		Cache:             options.Cache,
 		Replicate:         options.Replicate,
@@ -771,7 +786,7 @@ func (o *orbitDB) createStore(ctx context.Context, storeType string, parsedDBAdd
 		return nil, fmt.Errorf("unable to instantiate store: %w", err)
 	}
 
-	topic, err := o.pubsub.TopicSubscribe(ctx, parsedDBAddress.String())
+	topic, err := o.pubsub.TopicSubscribe(store.Context(), parsedDBAddress.String())
 	if err != nil {
 		return nil, fmt.Errorf("unable to subscribe to pubsub: %w", err)
 	}
@@ -782,11 +797,11 @@ func (o *orbitDB) createStore(ctx context.Context, storeType string, parsedDBAdd
 	// this is what hooks us into the message propagation layer
 	// and the p2p network
 	if *options.Replicate {
-		if err := o.storeListener(ctx, store, topic); err != nil {
+		if err := o.storeListener(store.Context(), store, topic); err != nil {
 			return nil, fmt.Errorf("unable to store listener: %w", err)
 		}
 
-		if err := o.pubSubChanListener(ctx, store, topic, parsedDBAddress); err != nil {
+		if err := o.pubSubChanListener(store.Context(), store, topic, parsedDBAddress); err != nil {
 			return nil, fmt.Errorf("unable to listen on pubsub: %w", err)
 		}
 	}
@@ -910,28 +925,28 @@ func (o *orbitDB) onNewPeerJoined(ctx context.Context, p peer.ID, store Store) {
 		o.logger.Debug(fmt.Sprintf("New peer '%s' connected to %s", p, store.Address().String()))
 	}
 
-	if err := o.exchangeHeads(ctx, p, store); err != nil {
+	if err := o.exchangeHeads(p, store); err != nil {
 		if !errors.Is(err, context.Canceled) {
 			o.logger.Error("unable to exchange heads", zap.Error(err))
 		}
 		return
 	}
 }
 
-func (o *orbitDB) exchangeHeads(ctx context.Context, p peer.ID, store Store) error {
+func (o *orbitDB) exchangeHeads(p peer.ID, store Store) error {
 	o.logger.Debug(fmt.Sprintf("connecting to %s", p))
-	if err := o.directChannel.Connect(ctx, p); err != nil {
+	if err := o.directChannel.Connect(store.Context(), p); err != nil {
 		return fmt.Errorf("unable to connect to peer: %w", err)
 	}
 	o.logger.Debug(fmt.Sprintf("connected to %s", p))
 
-	rawLocalHeads, err := store.Cache().Get(ctx, datastore.NewKey("_localHeads"))
+	rawLocalHeads, err := store.Cache().Get(store.Context(), datastore.NewKey("_localHeads"))
 	if err != nil && err != datastore.ErrNotFound {
 		return fmt.Errorf("unable to get local heads from cache: %w", err)
 	}
 
 	// @FIXME(gfanton): looks like activate this break the exchange
-	// rawRemoteHeads, err := store.Cache().Get(ctx, datastore.NewKey("_remoteHeads"))
+	// rawRemoteHeads, err := store.Cache().Get(store.Context(), datastore.NewKey("_remoteHeads"))
 	// if err != nil && err != datastore.ErrNotFound {
 	// 	return fmt.Errorf("unable to get data from cache: %w", err)
 	// }
@@ -962,7 +977,7 @@ func (o *orbitDB) exchangeHeads(ctx context.Context, p peer.ID, store Store) err
 
 	payloadstring := fmt.Sprintf("%x", payload)
 	o.logger.Debug("sending payload", zap.String("payload_string", payloadstring), zap.Any("payload", payload))
-	if err = o.directChannel.Send(ctx, p, payload); err != nil {
+	if err = o.directChannel.Send(store.Context(), p, payload); err != nil {
 		return fmt.Errorf("unable to send heads on direct channel: %w", err)
 	}
 
diff --git a/iface/interface.go b/iface/interface.go
@@ -181,6 +181,9 @@ type Store interface {
 	// Close Closes the store
 	Close() error
 
+	// IsClosed returns true if the store is closed
+	IsClosed() bool
+
 	// Address Returns the address for the current store
 	Address() address.Address
 
@@ -242,6 +245,9 @@ type Store interface {
 
 	// subscribe to events on this store
 	EventBus() event.Bus
+
+	// Get the context of the store
+	Context() context.Context
 }
 
 // EventLogStore A type of store that provides an append only log
@@ -371,7 +377,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
diff --git a/pubsub/oneonone/channel.go b/pubsub/oneonone/channel.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:
diff --git a/pubsub/pubsubraw/pubsub.go b/pubsub/pubsubraw/pubsub.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 {
diff --git a/stores/basestore/base_store.go b/stores/basestore/base_store.go
@@ -70,6 +70,8 @@ type BaseStore struct {
 	sortFn    ipfslog.SortFn
 	logger    *zap.Logger
 	tracer    trace.Tracer
+	ctx       context.Context
+	cancel    context.CancelFunc
 
 	// Deprecated: if possible don't use this, use EventBus() directly instead
 	events.EventEmitter
@@ -125,16 +127,31 @@ func (b *BaseStore) EventBus() event.Bus {
 	return b.options.EventBus
 }
 
+func (b *BaseStore) Context() context.Context {
+	return b.ctx
+}
+
+func (b *BaseStore) IsClosed() bool {
+	select {
+	case <-b.ctx.Done():
+		return true
+	default:
+		return false
+	}
+}
+
 // InitBaseStore Initializes the store base
-func (b *BaseStore) InitBaseStore(ctx context.Context, ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) error {
+func (b *BaseStore) InitBaseStore(ipfs coreapi.CoreAPI, identity *identityprovider.Identity, addr address.Address, options *iface.NewStoreOptions) error {
 	var err error
 
+	b.ctx, b.cancel = context.WithCancel(context.Background())
+
 	if options == nil {
 		options = &iface.NewStoreOptions{}
 	}
 
 	if options.EventBus == nil {
-		options.EventBus = b.EventEmitter.GetBus()
+		options.EventBus = eventbus.NewBus()
 	} else if err := b.SetBus(options.EventBus); err != nil {
 		return fmt.Errorf("unable set event bus: %w", err)
 	}
@@ -165,7 +182,7 @@ func (b *BaseStore) InitBaseStore(ctx context.Context, ipfs coreapi.CoreAPI, ide
 	} else {
 		manifestParams := accesscontroller.NewManifestParams(cid.Cid{}, true, "simple")
 		manifestParams.SetAccess("write", []string{identity.ID})
-		b.access, err = simple.NewSimpleAccessController(ctx, nil, manifestParams)
+		b.access, err = simple.NewSimpleAccessController(b.ctx, nil, manifestParams)
 
 		if err != nil {
 			return fmt.Errorf("unable to create simple access controller: %w", err)
@@ -202,8 +219,9 @@ func (b *BaseStore) InitBaseStore(ctx context.Context, ipfs coreapi.CoreAPI, ide
 	b.muIndex.Unlock()
 
 	b.replicator, err = replicator.NewReplicator(b, options.ReplicationConcurrency, &replicator.Options{
-		Logger: b.logger,
-		Tracer: b.tracer,
+		Logger:   b.logger,
+		EventBus: options.EventBus,
+		Tracer:   b.tracer,
 	})
 	if err != nil {
 		return fmt.Errorf("unable to init error: %w", err)
@@ -235,7 +253,7 @@ func (b *BaseStore) InitBaseStore(ctx context.Context, ipfs coreapi.CoreAPI, ide
 
 	go func() {
 		defer sub.Close()
-		ctx, span := b.tracer.Start(ctx, "base-store-main-loop", trace.WithAttributes(otkv.String("store-address", b.Address().String())))
+		ctx, span := b.tracer.Start(b.ctx, "base-store-main-loop", trace.WithAttributes(otkv.String("store-address", b.Address().String())))
 		defer span.End()
 
 		var e interface{}
@@ -305,6 +323,10 @@ func (b *BaseStore) InitBaseStore(ctx context.Context, ipfs coreapi.CoreAPI, ide
 }
 
 func (b *BaseStore) Close() error {
+	if b.IsClosed() {
+		return nil
+	}
+
 	// Replicator teardown logic
 	b.Replicator().Stop()
 
@@ -328,6 +350,8 @@ func (b *BaseStore) Close() error {
 		return fmt.Errorf("unable to close cache: %w", err)
 	}
 
+	b.cancel()
+
 	return nil
 }
 
diff --git a/stores/documentstore/document.go b/stores/documentstore/document.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)
 	}
diff --git a/stores/eventlogstore/log.go b/stores/eventlogstore/log.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)
 	}
diff --git a/stores/kvstore/keyvalue.go b/stores/kvstore/keyvalue.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)
 	}

Original file line number	Diff line number	Diff line change
`@@ -242,7 +242,7 @@ func DefaultStoreOptsForMap(keyField string) *iface.CreateDocumentDBOptions {`
`242`	`242`	`}`
`243`	`243`
`244`	`244`	`// NewOrbitDBDocumentStore Instantiates a new DocumentStore`
`245`		`-func NewOrbitDBDocumentStore(ctx context.Context, ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (iface.Store, error) {`
	`245`	`+func NewOrbitDBDocumentStore(ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (iface.Store, error) {`
`246`	`246`	`if options.StoreSpecificOpts == nil {`
`247`	`247`	`options.StoreSpecificOpts = DefaultStoreOptsForMap("_id")`
`248`	`248`	`}`
`@@ -271,7 +271,7 @@ func NewOrbitDBDocumentStore(ctx context.Context, ipfs coreapi.CoreAPI, identity`
`271`	`271`	`store := &orbitDBDocumentStore{docOpts: docOpts}`
`272`	`272`	`options.Index = func(_ []byte) iface.StoreIndex { return newDocumentIndex(docOpts) }`
`273`	`273`
`274`		`- err := store.InitBaseStore(ctx, ipfs, identity, addr, options)`
	`274`	`+ err := store.InitBaseStore(ipfs, identity, addr, options)`
`275`	`275`	`if err != nil {`
`276`	`276`	`return nil, fmt.Errorf("unable to initialize document store: %w", err)`
`277`	`277`	`}`
Original file line number	Diff line number	Diff line change
`@@ -201,11 +201,12 @@ func (o *orbitDBEventLogStore) Type() string {`
`201`	`201`	`}`
`202`	`202`
`203`	`203`	`// NewOrbitDBEventLogStore Instantiates a new EventLogStore`
`204`		`-func NewOrbitDBEventLogStore(ctx context.Context, ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (i iface.Store, e error) {`
	`204`	`+func NewOrbitDBEventLogStore(ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (i iface.Store, e error) {`
	`205`	`+`
`205`	`206`	`store := &orbitDBEventLogStore{}`
`206`	`207`	`options.Index = NewEventIndex`
`207`	`208`
`208`		`- err := store.InitBaseStore(ctx, ipfs, identity, addr, options)`
	`209`	`+ err := store.InitBaseStore(ipfs, identity, addr, options)`
`209`	`210`	`if err != nil {`
`210`	`211`	`return nil, fmt.Errorf("unable to initialize base store: %w", err)`
`211`	`212`	`}`
Original file line number	Diff line number	Diff line change
`@@ -85,12 +85,12 @@ func (o *orbitDBKeyValue) Type() string {`
`85`	`85`	`}`
`86`	`86`
`87`	`87`	`// NewOrbitDBKeyValue Instantiates a new KeyValueStore`
`88`		`-func NewOrbitDBKeyValue(ctx context.Context, ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (i iface.Store, e error) {`
	`88`	`+func NewOrbitDBKeyValue(ipfs coreapi.CoreAPI, identity identityprovider.Identity, addr address.Address, options iface.NewStoreOptions) (i iface.Store, e error) {`
`89`	`89`	`store := &orbitDBKeyValue{}`
`90`	`90`
`91`	`91`	`options.Index = NewKVIndex`
`92`	`92`
`93`		`- err := store.InitBaseStore(ctx, ipfs, identity, addr, options)`
	`93`	`+ err := store.InitBaseStore(ipfs, identity, addr, options)`
`94`	`94`	`if err != nil {`
`95`	`95`	`return nil, fmt.Errorf("unable to initialize base store: %w", err)`
`96`	`96`	`}`