diff --git a/nodedb.go b/nodedb.go index 3cd24324b..3df268ddd 100644 --- a/nodedb.go +++ b/nodedb.go @@ -442,31 +442,37 @@ func (rkc *rootkeyCache) setRootKey(version int64, rootKey []byte) { // deletes orphans func (ndb *nodeDB) deleteVersion(version int64, cache *rootkeyCache) error { rootKey, err := cache.getRootKey(ndb, version) - if err != nil { + if err != nil && !errors.Is(err, ErrVersionDoesNotExist) { return err } - if err := ndb.traverseOrphans(version, version+1, func(orphan *Node) error { - if orphan.nodeKey.nonce == 0 && !orphan.isLegacy { - // if the orphan is a reformatted root, it can be a legacy root - // so it should be removed from the pruning process. - if err := ndb.deleteFromPruning(ndb.legacyNodeKey(orphan.hash)); err != nil { - return err + if errors.Is(err, ErrVersionDoesNotExist) { + ndb.logger.Error("Error while pruning, moving on the the next version in the store", "version missing", version, "next version", version+1, "err", err) + } + + if rootKey != nil { + if err := ndb.traverseOrphans(version, version+1, func(orphan *Node) error { + if orphan.nodeKey.nonce == 0 && !orphan.isLegacy { + // if the orphan is a reformatted root, it can be a legacy root + // so it should be removed from the pruning process. + if err := ndb.deleteFromPruning(ndb.legacyNodeKey(orphan.hash)); err != nil { + return err + } } + if orphan.nodeKey.nonce == 1 && orphan.nodeKey.version < version { + // if the orphan is referred to the previous root, it should be reformatted + // to (version, 0), because the root (version, 1) should be removed but not + // applied now due to the batch writing. + orphan.nodeKey.nonce = 0 + } + nk := orphan.GetKey() + if orphan.isLegacy { + return ndb.deleteFromPruning(ndb.legacyNodeKey(nk)) + } + return ndb.deleteFromPruning(ndb.nodeKey(nk)) + }); err != nil && err != ErrVersionDoesNotExist { + return err } - if orphan.nodeKey.nonce == 1 && orphan.nodeKey.version < version { - // if the orphan is referred to the previous root, it should be reformatted - // to (version, 0), because the root (version, 1) should be removed but not - // applied now due to the batch writing. - orphan.nodeKey.nonce = 0 - } - nk := orphan.GetKey() - if orphan.isLegacy { - return ndb.deleteFromPruning(ndb.legacyNodeKey(nk)) - } - return ndb.deleteFromPruning(ndb.nodeKey(nk)) - }); err != nil { - return err } literalRootKey := GetRootKey(version) @@ -480,7 +486,7 @@ func (ndb *nodeDB) deleteVersion(version int64, cache *rootkeyCache) error { // check if the version is referred by the next version nextRootKey, err := cache.getRootKey(ndb, version+1) - if err != nil { + if err != nil && !errors.Is(err, ErrVersionDoesNotExist) { return err } if bytes.Equal(literalRootKey, nextRootKey) { @@ -703,7 +709,15 @@ func (ndb *nodeDB) deleteVersionsTo(toVersion int64) error { if err := ndb.deleteLegacyVersions(legacyLatestVersion); err != nil { ndb.logger.Error("Error deleting legacy versions", "err", err) } - first = legacyLatestVersion + 1 + // NOTE: When pruning is broken for legacy versions we need to find the + // latest non legacy version in the store + // TODO: Make sure legacy pruning works as expected and does not fail + firstNonLegacyVersion, err := ndb.getFirstNonLegacyVersion() + if err != nil { + return err + } + first = firstNonLegacyVersion + // reset the legacy latest version forcibly to avoid multiple calls ndb.resetLegacyLatestVersion(-1) } @@ -745,6 +759,35 @@ func (ndb *nodeDB) legacyRootKey(version int64) []byte { return legacyRootKeyFormat.Key(version) } +// getFirstNonLegacyVersion binary searches the store for the first non-legacy version +func (ndb *nodeDB) getFirstNonLegacyVersion() (int64, error) { + ndb.mtx.Lock() + firstVersion := ndb.firstVersion + ndb.mtx.Unlock() + + // Find the first version + latestVersion, err := ndb.getLatestVersion() + if err != nil { + return 0, err + } + for firstVersion < latestVersion { + version := (latestVersion + firstVersion) >> 1 + has, err := ndb.hasVersion(version) + if err != nil { + return 0, err + } + if has { + latestVersion = version + } else { + firstVersion = version + 1 + } + } + + ndb.resetFirstVersion(latestVersion) + + return latestVersion, nil +} + func (ndb *nodeDB) getFirstVersion() (int64, error) { ndb.mtx.Lock() firstVersion := ndb.firstVersion