[#211] engine: Unify shard mode checks for tree operations
All checks were successful
ci/woodpecker/pr/pre-commit Pipeline was successful

All operations must ensure the shard is not in a degraded mode.
Write operations must also ensure the shard is not in a read-only mode.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
Evgenii Stratonikov 2023-04-05 10:46:59 +03:00
parent 48e7100556
commit 731979a482

View file

@ -23,6 +23,9 @@ func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Mo
if s.info.Mode.ReadOnly() { if s.info.Mode.ReadOnly() {
return nil, ErrReadOnlyMode return nil, ErrReadOnlyMode
} }
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeMove(d, treeID, m) return s.pilorama.TreeMove(d, treeID, m)
} }
@ -38,6 +41,9 @@ func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr stri
if s.info.Mode.ReadOnly() { if s.info.Mode.ReadOnly() {
return nil, ErrReadOnlyMode return nil, ErrReadOnlyMode
} }
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta) return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta)
} }
@ -53,6 +59,9 @@ func (s *Shard) TreeApply(cnr cidSDK.ID, treeID string, m *pilorama.Move, backgr
if s.info.Mode.ReadOnly() { if s.info.Mode.ReadOnly() {
return ErrReadOnlyMode return ErrReadOnlyMode
} }
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeApply(cnr, treeID, m, backgroundSync) return s.pilorama.TreeApply(cnr, treeID, m, backgroundSync)
} }
@ -61,6 +70,13 @@ func (s *Shard) TreeGetByPath(cid cidSDK.ID, treeID string, attr string, path []
if s.pilorama == nil { if s.pilorama == nil {
return nil, ErrPiloramaDisabled return nil, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeGetByPath(cid, treeID, attr, path, latest) return s.pilorama.TreeGetByPath(cid, treeID, attr, path, latest)
} }
@ -69,6 +85,13 @@ func (s *Shard) TreeGetMeta(cid cidSDK.ID, treeID string, nodeID pilorama.Node)
if s.pilorama == nil { if s.pilorama == nil {
return pilorama.Meta{}, 0, ErrPiloramaDisabled return pilorama.Meta{}, 0, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return pilorama.Meta{}, 0, ErrDegradedMode
}
return s.pilorama.TreeGetMeta(cid, treeID, nodeID) return s.pilorama.TreeGetMeta(cid, treeID, nodeID)
} }
@ -77,6 +100,13 @@ func (s *Shard) TreeGetChildren(cid cidSDK.ID, treeID string, nodeID pilorama.No
if s.pilorama == nil { if s.pilorama == nil {
return nil, ErrPiloramaDisabled return nil, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeGetChildren(cid, treeID, nodeID) return s.pilorama.TreeGetChildren(cid, treeID, nodeID)
} }
@ -85,6 +115,13 @@ func (s *Shard) TreeGetOpLog(cid cidSDK.ID, treeID string, height uint64) (pilor
if s.pilorama == nil { if s.pilorama == nil {
return pilorama.Move{}, ErrPiloramaDisabled return pilorama.Move{}, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return pilorama.Move{}, ErrDegradedMode
}
return s.pilorama.TreeGetOpLog(cid, treeID, height) return s.pilorama.TreeGetOpLog(cid, treeID, height)
} }
@ -93,6 +130,13 @@ func (s *Shard) TreeDrop(cid cidSDK.ID, treeID string) error {
if s.pilorama == nil { if s.pilorama == nil {
return ErrPiloramaDisabled return ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeDrop(cid, treeID) return s.pilorama.TreeDrop(cid, treeID)
} }
@ -101,6 +145,13 @@ func (s *Shard) TreeList(cid cidSDK.ID) ([]string, error) {
if s.pilorama == nil { if s.pilorama == nil {
return nil, ErrPiloramaDisabled return nil, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}
return s.pilorama.TreeList(cid) return s.pilorama.TreeList(cid)
} }
@ -109,6 +160,13 @@ func (s *Shard) TreeExists(cid cidSDK.ID, treeID string) (bool, error) {
if s.pilorama == nil { if s.pilorama == nil {
return false, ErrPiloramaDisabled return false, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return false, ErrDegradedMode
}
return s.pilorama.TreeExists(cid, treeID) return s.pilorama.TreeExists(cid, treeID)
} }
@ -117,6 +175,16 @@ func (s *Shard) TreeUpdateLastSyncHeight(cid cidSDK.ID, treeID string, height ui
if s.pilorama == nil { if s.pilorama == nil {
return ErrPiloramaDisabled return ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.ReadOnly() {
return ErrReadOnlyMode
}
if s.info.Mode.NoMetabase() {
return ErrDegradedMode
}
return s.pilorama.TreeUpdateLastSyncHeight(cid, treeID, height) return s.pilorama.TreeUpdateLastSyncHeight(cid, treeID, height)
} }
@ -125,5 +193,12 @@ func (s *Shard) TreeLastSyncHeight(cid cidSDK.ID, treeID string) (uint64, error)
if s.pilorama == nil { if s.pilorama == nil {
return 0, ErrPiloramaDisabled return 0, ErrPiloramaDisabled
} }
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return 0, ErrDegradedMode
}
return s.pilorama.TreeLastSyncHeight(cid, treeID) return s.pilorama.TreeLastSyncHeight(cid, treeID)
} }