TrueCloudLab/frostfs-node
21
1
Fork 27
Code Issues 107 Pull requests 5 Projects Releases 2 Activity Actions

[#1437] blobovnicza: Fix contextcheck linter

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2024-10-21 11:20:17 +03:00
parent 692749dba1
commit fa0cb3448e
Signed by: dstepanov-yadro
GPG key ID: 237AF1A763293BC0
22 changed files with 82 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/frostfs-lens/internal/blobovnicza/root.go
View file

@ -27,7 +27,7 @@ func openBlobovnicza(cmd *cobra.Command) *blobovnicza.Blobovnicza {
blobovnicza.WithPath(vPath),
blobovnicza.WithReadOnly(true),
)
common.ExitOnErr(cmd, blz.Open())
common.ExitOnErr(cmd, blz.Open(cmd.Context()))
return blz
}

2
pkg/local_object_storage/blobovnicza/blobovnicza_test.go
View file

@ -69,7 +69,7 @@ func TestBlobovnicza(t *testing.T) {
defer os.Remove(p)
// open Blobovnicza
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
// initialize Blobovnicza
require.NoError(t, blz.Init())

6
pkg/local_object_storage/blobovnicza/control.go
View file

@ -16,7 +16,7 @@ import (
//
// If the database file does not exist, it will be created automatically.
// If blobovnicza is already open, does nothing.
func (b *Blobovnicza) Open() error {
func (b *Blobovnicza) Open(ctx context.Context) error {
b.controlMtx.Lock()
defer b.controlMtx.Unlock()
@ -24,7 +24,7 @@ func (b *Blobovnicza) Open() error {
return nil
}
b.log.Debug(context.Background(), logs.BlobovniczaCreatingDirectoryForBoltDB,
b.log.Debug(ctx, logs.BlobovniczaCreatingDirectoryForBoltDB,
zap.String("path", b.path),
zap.Bool("ro", b.boltOptions.ReadOnly),
)
@ -38,7 +38,7 @@ func (b *Blobovnicza) Open() error {
}
}
b.log.Debug(context.Background(), logs.BlobovniczaOpeningBoltDB,
b.log.Debug(ctx, logs.BlobovniczaOpeningBoltDB,
zap.String("path", b.path),
zap.Stringer("permissions", b.perm),
)

2
pkg/local_object_storage/blobovnicza/get_test.go
View file

@ -26,7 +26,7 @@ func TestBlobovnicza_Get(t *testing.T) {
WithObjectSizeLimit(szLimit),
)
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
}

2
pkg/local_object_storage/blobovnicza/iterate_test.go
View file

@ -15,7 +15,7 @@ import (
func TestBlobovniczaIterate(t *testing.T) {
filename := filepath.Join(t.TempDir(), "blob")
b := New(WithPath(filename))
require.NoError(t, b.Open())
require.NoError(t, b.Open(context.Background()))
require.NoError(t, b.Init())
data := [][]byte{{0, 1, 2, 3}, {5, 6, 7, 8}}

23
pkg/local_object_storage/blobstor/blobovniczatree/active.go
View file

@ -1,6 +1,7 @@
package blobovniczatree
import (
"context"
"path/filepath"
"sync"
@ -53,8 +54,8 @@ func newActiveDBManager(dbManager *dbManager, rootPath string) *activeDBManager
// GetOpenedActiveDBForLevel returns active DB for level.
// DB must be closed after use.
func (m *activeDBManager) GetOpenedActiveDBForLevel(lvlPath string) (*activeDB, error) {
activeDB, err := m.getCurrentActiveIfOk(lvlPath)
func (m *activeDBManager) GetOpenedActiveDBForLevel(ctx context.Context, lvlPath string) (*activeDB, error) {
activeDB, err := m.getCurrentActiveIfOk(ctx, lvlPath)
if err != nil {
return nil, err
}
@ -62,7 +63,7 @@ func (m *activeDBManager) GetOpenedActiveDBForLevel(lvlPath string) (*activeDB,
return activeDB, nil
}
return m.updateAndGetActive(lvlPath)
return m.updateAndGetActive(ctx, lvlPath)
}
func (m *activeDBManager) Open() {
@ -83,7 +84,7 @@ func (m *activeDBManager) Close() {
m.closed = true
}
func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error) {
func (m *activeDBManager) getCurrentActiveIfOk(ctx context.Context, lvlPath string) (*activeDB, error) {
m.levelToActiveDBGuard.RLock()
defer m.levelToActiveDBGuard.RUnlock()
@ -96,7 +97,7 @@ func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error
return nil, nil
}
blz, err := db.Open() // open db for usage, will be closed on activeDB.Close()
blz, err := db.Open(ctx) // open db for usage, will be closed on activeDB.Close()
if err != nil {
return nil, err
}
@ -112,11 +113,11 @@ func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error
}, nil
}
func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error) {
func (m *activeDBManager) updateAndGetActive(ctx context.Context, lvlPath string) (*activeDB, error) {
m.levelLock.Lock(lvlPath)
defer m.levelLock.Unlock(lvlPath)
current, err := m.getCurrentActiveIfOk(lvlPath)
current, err := m.getCurrentActiveIfOk(ctx, lvlPath)
if err != nil {
return nil, err
}
@ -124,7 +125,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
return current, nil
}
nextShDB, err := m.getNextSharedDB(lvlPath)
nextShDB, err := m.getNextSharedDB(ctx, lvlPath)
if err != nil {
return nil, err
}
@ -133,7 +134,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
return nil, nil
}
blz, err := nextShDB.Open() // open db for client, client must call Close() after usage
blz, err := nextShDB.Open(ctx) // open db for client, client must call Close() after usage
if err != nil {
return nil, err
}
@ -143,7 +144,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
}, nil
}
func (m *activeDBManager) getNextSharedDB(lvlPath string) (*sharedDB, error) {
func (m *activeDBManager) getNextSharedDB(ctx context.Context, lvlPath string) (*sharedDB, error) {
var nextActiveDBIdx uint64
hasActive, currentIdx := m.hasActiveDB(lvlPath)
if hasActive {
@ -160,7 +161,7 @@ func (m *activeDBManager) getNextSharedDB(lvlPath string) (*sharedDB, error) {
path := filepath.Join(lvlPath, u64ToHexStringExt(nextActiveDBIdx))
next := m.dbManager.GetByPath(path)
_, err := next.Open() // open db to hold active DB open, will be closed if db is full, after m.replace or by activeDBManager.Close()
_, err := next.Open(ctx) // open db to hold active DB open, will be closed if db is full, after m.replace or by activeDBManager.Close()
if err != nil {
return nil, err
}

8
pkg/local_object_storage/blobstor/blobovniczatree/cache.go
View file

@ -81,12 +81,12 @@ func (c *dbCache) Close() {
c.closed = true
}
func (c *dbCache) GetOrCreate(path string) *sharedDB {
func (c *dbCache) GetOrCreate(ctx context.Context, path string) *sharedDB {
value := c.getExisted(path)
if value != nil {
return value
}
return c.create(path)
return c.create(ctx, path)
}
func (c *dbCache) EvictAndMarkNonCached(path string) {
@ -122,7 +122,7 @@ func (c *dbCache) getExisted(path string) *sharedDB {
return nil
}
func (c *dbCache) create(path string) *sharedDB {
func (c *dbCache) create(ctx context.Context, path string) *sharedDB {
c.pathLock.Lock(path)
defer c.pathLock.Unlock(path)
@ -133,7 +133,7 @@ func (c *dbCache) create(path string) *sharedDB {
value = c.dbManager.GetByPath(path)
_, err := value.Open() // open db to hold reference, closed by evictedDB.Close() or if cache closed
_, err := value.Open(ctx) // open db to hold reference, closed by evictedDB.Close() or if cache closed
if err != nil {
return value
}

6
pkg/local_object_storage/blobstor/blobovniczatree/control.go
View file

@ -46,7 +46,7 @@ func (b *Blobovniczas) initializeDBs(ctx context.Context) error {
eg.Go(func() error {
p = strings.TrimSuffix(p, rebuildSuffix)
shBlz := b.getBlobovniczaWithoutCaching(p)
blz, err := shBlz.Open()
blz, err := shBlz.Open(egCtx)
if err != nil {
return err
}
@ -91,8 +91,8 @@ func (b *Blobovniczas) Close() error {
// returns blobovnicza with path p
//
// If blobovnicza is already cached, instance from cache is returned w/o changes.
func (b *Blobovniczas) getBlobovnicza(p string) *sharedDB {
return b.dbCache.GetOrCreate(p)
func (b *Blobovniczas) getBlobovnicza(ctx context.Context, p string) *sharedDB {
return b.dbCache.GetOrCreate(ctx, p)
}
func (b *Blobovniczas) getBlobovniczaWithoutCaching(p string) *sharedDB {

4
pkg/local_object_storage/blobstor/blobovniczatree/count.go
View file

@ -16,13 +16,13 @@ func (b *Blobovniczas) ObjectsCount(ctx context.Context) (uint64, error) {
b.metrics.ObjectsCount(time.Since(startedAt), success)
}()
_, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.ObjectsCount")
ctx, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.ObjectsCount")
defer span.End()
var result uint64
err := b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
shDB := b.getBlobovniczaWithoutCaching(p)
blz, err := shDB.Open()
blz, err := shDB.Open(ctx)
if err != nil {
return true, err
}

8
pkg/local_object_storage/blobstor/blobovniczatree/delete.go
View file

@ -61,8 +61,8 @@ func (b *Blobovniczas) Delete(ctx context.Context, prm common.DeletePrm) (res co
if prm.StorageID != nil {
id := NewIDFromBytes(prm.StorageID)
shBlz := b.getBlobovnicza(id.Path())
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, id.Path())
blz, err := shBlz.Open(ctx)
if err != nil {
return res, err
}
@ -109,8 +109,8 @@ func (b *Blobovniczas) Delete(ctx context.Context, prm common.DeletePrm) (res co
//
// returns no error if object was removed from some blobovnicza of the same level.
func (b *Blobovniczas) deleteObjectFromLevel(ctx context.Context, prm blobovnicza.DeletePrm, blzPath string) (common.DeleteRes, error) {
shBlz := b.getBlobovnicza(blzPath)
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, blzPath)
blz, err := shBlz.Open(ctx)
if err != nil {
return common.DeleteRes{}, err
}

4
pkg/local_object_storage/blobstor/blobovniczatree/exists.go
View file

@ -37,8 +37,8 @@ func (b *Blobovniczas) Exists(ctx context.Context, prm common.ExistsPrm) (common
if prm.StorageID != nil {
id := NewIDFromBytes(prm.StorageID)
shBlz := b.getBlobovnicza(id.Path())
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, id.Path())
blz, err := shBlz.Open(ctx)
if err != nil {
return common.ExistsRes{}, err
}

8
pkg/local_object_storage/blobstor/blobovniczatree/get.go
View file

@ -48,8 +48,8 @@ func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.G
if prm.StorageID != nil {
id := NewIDFromBytes(prm.StorageID)
shBlz := b.getBlobovnicza(id.Path())
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, id.Path())
blz, err := shBlz.Open(ctx)
if err != nil {
return res, err
}
@ -95,8 +95,8 @@ func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.G
// returns error if object could not be read from any blobovnicza of the same level.
func (b *Blobovniczas) getObjectFromLevel(ctx context.Context, prm blobovnicza.GetPrm, blzPath string) (common.GetRes, error) {
// open blobovnicza (cached inside)
shBlz := b.getBlobovnicza(blzPath)
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, blzPath)
blz, err := shBlz.Open(ctx)
if err != nil {
return common.GetRes{}, err
}

8
pkg/local_object_storage/blobstor/blobovniczatree/get_range.go
View file

@ -47,8 +47,8 @@ func (b *Blobovniczas) GetRange(ctx context.Context, prm common.GetRangePrm) (re
if prm.StorageID != nil {
id := NewIDFromBytes(prm.StorageID)
shBlz := b.getBlobovnicza(id.Path())
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, id.Path())
blz, err := shBlz.Open(ctx)
if err != nil {
return common.GetRangeRes{}, err
}
@ -103,8 +103,8 @@ func (b *Blobovniczas) GetRange(ctx context.Context, prm common.GetRangePrm) (re
// returns error if object could not be read from any blobovnicza of the same level.
func (b *Blobovniczas) getRangeFromLevel(ctx context.Context, prm common.GetRangePrm, blzPath string) (common.GetRangeRes, error) {
// open blobovnicza (cached inside)
shBlz := b.getBlobovnicza(blzPath)
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, blzPath)
blz, err := shBlz.Open(ctx)
if err != nil {
return common.GetRangeRes{}, err
}

4
pkg/local_object_storage/blobstor/blobovniczatree/iterate.go
View file

@ -72,8 +72,8 @@ func (b *Blobovniczas) Iterate(ctx context.Context, prm common.IteratePrm) (comm
// iterator over all Blobovniczas in unsorted order. Break on f's error return.
func (b *Blobovniczas) iterateBlobovniczas(ctx context.Context, ignoreErrors bool, f func(string, *blobovnicza.Blobovnicza) error) error {
return b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
shBlz := b.getBlobovnicza(p)
blz, err := shBlz.Open()
shBlz := b.getBlobovnicza(ctx, p)
blz, err := shBlz.Open(ctx)
if err != nil {
if ignoreErrors {
b.log.Warn(ctx, logs.BlobstorErrorOccurredDuringTheIteration,

4
pkg/local_object_storage/blobstor/blobovniczatree/manager.go
View file

@ -49,7 +49,7 @@ func newSharedDB(options []blobovnicza.Option, path string, readOnly bool,
}
}
func (b *sharedDB) Open() (*blobovnicza.Blobovnicza, error) {
func (b *sharedDB) Open(ctx context.Context) (*blobovnicza.Blobovnicza, error) {
if b.closedFlag.Load() {
return nil, errClosed
}
@ -68,7 +68,7 @@ func (b *sharedDB) Open() (*blobovnicza.Blobovnicza, error) {
blobovnicza.WithMetrics(b.metrics),
)...)
if err := blz.Open(); err != nil {
if err := blz.Open(ctx); err != nil {
return nil, fmt.Errorf("could not open blobovnicza %s: %w", b.path, err)
}
if err := blz.Init(); err != nil {

2
pkg/local_object_storage/blobstor/blobovniczatree/put.go
View file

@ -77,7 +77,7 @@ type putIterator struct {
}
func (i *putIterator) iterate(ctx context.Context, lvlPath string) (bool, error) {
active, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(lvlPath)
active, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(ctx, lvlPath)
if err != nil {
if !isLogical(err) {
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)

22
pkg/local_object_storage/blobstor/blobovniczatree/rebuild.go
View file

@ -165,7 +165,7 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe
continue
}
path := filepath.Join(lvlPath, e.Name())
resettlementRequired, err := b.rebuildBySize(path, target)
resettlementRequired, err := b.rebuildBySize(ctx, path, target)
if err != nil {
return false, err
}
@ -180,9 +180,9 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe
return result, nil
}
func (b *Blobovniczas) rebuildBySize(path string, targetFillPercent int) (bool, error) {
shDB := b.getBlobovnicza(path)
blz, err := shDB.Open()
func (b *Blobovniczas) rebuildBySize(ctx context.Context, path string, targetFillPercent int) (bool, error) {
shDB := b.getBlobovnicza(ctx, path)
blz, err := shDB.Open(ctx)
if err != nil {
return false, err
}
@ -196,8 +196,8 @@ func (b *Blobovniczas) rebuildBySize(path string, targetFillPercent int) (bool,
}
func (b *Blobovniczas) rebuildDB(ctx context.Context, path string, meta common.MetaStorage, limiter common.ConcurrentWorkersLimiter) (uint64, error) {
shDB := b.getBlobovnicza(path)
blz, err := shDB.Open()
shDB := b.getBlobovnicza(ctx, path)
blz, err := shDB.Open(ctx)
if err != nil {
return 0, err
}
@ -365,8 +365,8 @@ func (b *Blobovniczas) completeIncompletedMove(ctx context.Context, metaStore co
err := b.iterateIncompletedRebuildDBPaths(ctx, func(s string) (bool, error) {
rebuildTmpFilePath := s
s = strings.TrimSuffix(s, rebuildSuffix)
shDB := b.getBlobovnicza(s)
blz, err := shDB.Open()
shDB := b.getBlobovnicza(ctx, s)
blz, err := shDB.Open(ctx)
if err != nil {
return true, err
}
@ -398,8 +398,8 @@ func (b *Blobovniczas) completeIncompletedMove(ctx context.Context, metaStore co
func (b *Blobovniczas) performMove(ctx context.Context, source *blobovnicza.Blobovnicza, sourcePath string,
move blobovnicza.MoveInfo, metaStore common.MetaStorage,
) error {
targetDB := b.getBlobovnicza(NewIDFromBytes(move.TargetStorageID).Path())
target, err := targetDB.Open()
targetDB := b.getBlobovnicza(ctx, NewIDFromBytes(move.TargetStorageID).Path())
target, err := targetDB.Open(ctx)
if err != nil {
return err
}
@ -477,7 +477,7 @@ type moveIterator struct {
}
func (i *moveIterator) tryMoveToLvl(ctx context.Context, lvlPath string) (bool, error) {
target, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(lvlPath)
target, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(ctx, lvlPath)
if err != nil {
if !isLogical(err) {
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)

14
pkg/local_object_storage/blobstor/blobovniczatree/rebuild_failover_test.go
View file

@ -35,7 +35,7 @@ func testRebuildFailoverOnlyMoveInfoSaved(t *testing.T) {
dir := t.TempDir()
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
obj := blobstortest.NewObject(1024)
@ -65,7 +65,7 @@ func testRebuildFailoverObjectSavedToTarget(t *testing.T) {
dir := t.TempDir()
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
obj := blobstortest.NewObject(1024)
@ -89,7 +89,7 @@ func testRebuildFailoverObjectSavedToTarget(t *testing.T) {
require.NoError(t, err)
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
_, err = blz.Put(context.Background(), pPrm)
@ -105,7 +105,7 @@ func testRebuildFailoverObjectDeletedFromSource(t *testing.T) {
dir := t.TempDir()
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
obj := blobstortest.NewObject(1024)
@ -123,7 +123,7 @@ func testRebuildFailoverObjectDeletedFromSource(t *testing.T) {
require.NoError(t, err)
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
var pPrm blobovnicza.PutPrm
@ -173,7 +173,7 @@ func testRebuildFailoverValidate(t *testing.T, dir string, obj *objectSDK.Object
require.NoError(t, b.Close())
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
moveInfo, err := blz.ListMoveInfo(context.Background())
@ -188,7 +188,7 @@ func testRebuildFailoverValidate(t *testing.T, dir string, obj *objectSDK.Object
require.NoError(t, blz.Close())
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
require.NoError(t, blz.Open())
require.NoError(t, blz.Open(context.Background()))
require.NoError(t, blz.Init())
moveInfo, err = blz.ListMoveInfo(context.Background())

4
pkg/local_object_storage/engine/engine.go
View file

@ -269,8 +269,8 @@ type containerSource struct {
func (s *containerSource) IsContainerAvailable(ctx context.Context, id cid.ID) (bool, error) {
select {
case <-context.Background().Done():
return false, context.Background().Err()
case <-ctx.Done():
return false, ctx.Err()
default:
}

4
pkg/morph/client/client.go
View file

@ -388,7 +388,7 @@ func (c *Client) Wait(ctx context.Context, n uint32) error {
height, err = c.rpcActor.GetBlockCount()
if err != nil {
c.logger.Error(context.Background(), logs.ClientCantGetBlockchainHeight,
c.logger.Error(ctx, logs.ClientCantGetBlockchainHeight,
zap.String("error", err.Error()))
return nil
}
@ -402,7 +402,7 @@ func (c *Client) Wait(ctx context.Context, n uint32) error {
newHeight, err = c.rpcActor.GetBlockCount()
if err != nil {
c.logger.Error(context.Background(), logs.ClientCantGetBlockchainHeight243,
c.logger.Error(ctx, logs.ClientCantGetBlockchainHeight243,
zap.String("error", err.Error()))
return nil
}

16
pkg/morph/event/listener.go
View file

@ -269,7 +269,7 @@ loop:
continue loop
}
l.handleNotifyEvent(notifyEvent)
l.handleNotifyEvent(ctx, notifyEvent)
case notaryEvent, ok := <-chs.NotaryRequestsCh:
if !ok {
l.log.Warn(ctx, logs.EventStopEventListenerByNotaryChannel)
@ -316,16 +316,16 @@ func (l *listener) handleNotaryEvent(notaryEvent *result.NotaryRequestEvent) {
}
}
func (l *listener) handleNotifyEvent(notifyEvent *state.ContainedNotificationEvent) {
func (l *listener) handleNotifyEvent(ctx context.Context, notifyEvent *state.ContainedNotificationEvent) {
if err := l.pool.Submit(func() {
l.parseAndHandleNotification(notifyEvent)
l.parseAndHandleNotification(ctx, notifyEvent)
}); err != nil {
l.log.Warn(context.Background(), logs.EventListenerWorkerPoolDrained,
l.log.Warn(ctx, logs.EventListenerWorkerPoolDrained,
zap.Int("capacity", l.pool.Cap()))
}
}
func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotificationEvent) {
func (l *listener) parseAndHandleNotification(ctx context.Context, notifyEvent *state.ContainedNotificationEvent) {
log := l.log.With(
zap.String("script hash LE", notifyEvent.ScriptHash.StringLE()),
)
@ -347,7 +347,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
l.mtx.RUnlock()
if !ok {
log.Debug(context.Background(), logs.EventEventParserNotSet)
log.Debug(ctx, logs.EventEventParserNotSet)
return
}
@ -355,7 +355,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
// parse the notification event
event, err := parser(notifyEvent)
if err != nil {
log.Warn(context.Background(), logs.EventCouldNotParseNotificationEvent,
log.Warn(ctx, logs.EventCouldNotParseNotificationEvent,
zap.String("error", err.Error()),
)
@ -368,7 +368,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
l.mtx.RUnlock()
if len(handlers) == 0 {
log.Info(context.Background(), logs.EventNotificationHandlersForParsedNotificationEventWereNotRegistered,
log.Info(ctx, logs.EventNotificationHandlersForParsedNotificationEventWereNotRegistered,
zap.Any("event", event),
)

10
pkg/morph/subscriber/subscriber.go
View file

@ -254,7 +254,7 @@ func (s *subscriber) switchEndpoint(ctx context.Context, finishCh chan<- bool) b
s.Lock()
chs := newSubChannels()
go func() {
finishCh <- s.restoreSubscriptions(chs.NotifyChan, chs.BlockChan, chs.NotaryChan)
finishCh <- s.restoreSubscriptions(ctx, chs.NotifyChan, chs.BlockChan, chs.NotaryChan)
}()
s.current = chs
s.Unlock()
@ -295,7 +295,7 @@ drainloop:
// restoreSubscriptions restores subscriptions according to
// cached information about them.
func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotificationEvent,
func (s *subscriber) restoreSubscriptions(ctx context.Context, notifCh chan<- *state.ContainedNotificationEvent,
blCh chan<- *block.Block, notaryCh chan<- *result.NotaryRequestEvent,
) bool {
var err error
@ -304,7 +304,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
if s.subscribedToNewBlocks {
_, err = s.client.ReceiveBlocks(blCh)
if err != nil {
s.log.Error(context.Background(), logs.ClientCouldNotRestoreBlockSubscriptionAfterRPCSwitch, zap.Error(err))
s.log.Error(ctx, logs.ClientCouldNotRestoreBlockSubscriptionAfterRPCSwitch, zap.Error(err))
return false
}
}
@ -313,7 +313,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
for contract := range s.subscribedEvents {
_, err = s.client.ReceiveExecutionNotifications(contract, notifCh)
if err != nil {
s.log.Error(context.Background(), logs.ClientCouldNotRestoreNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
s.log.Error(ctx, logs.ClientCouldNotRestoreNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
return false
}
}
@ -322,7 +322,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
for signer := range s.subscribedNotaryEvents {
_, err = s.client.ReceiveNotaryRequests(signer, notaryCh)
if err != nil {
s.log.Error(context.Background(), logs.ClientCouldNotRestoreNotaryNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
s.log.Error(ctx, logs.ClientCouldNotRestoreNotaryNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
return false
}
}