Frame pruning fixes (#405)

* Consistently close iterators

* Prune frames in batches

* Add compact on start flag

node/consensus/data/frame_pruner.go

@@ -1,13 +0,0 @@
-package data
-
-import "go.uber.org/zap"
-
-func (e *DataClockConsensusEngine) pruneFrames(maxFrame uint64) error {
-	e.logger.Info("pruning frames", zap.Uint64("max_frame_to_prune", maxFrame))
-	err := e.clockStore.DeleteDataClockFrameRange(e.filter, 1, maxFrame)
-	if err != nil {
-		e.logger.Error("failed to prune frames", zap.Error(err))
-		return err
-	}
-	return nil
-}

node/consensus/data/main_data_loop.go

@@ -103,6 +103,10 @@ func (e *DataClockConsensusEngine) runFramePruning() {
 
 	e.logger.Info("frame pruning enabled, waiting for delay timeout expiry")
 
+	from := uint64(1)
+	maxFrames := uint64(e.config.Engine.MaxFrames)
+	batchSize := uint64(1000)
+outer:
 	for {
 		select {
 		case <-e.ctx.Done():
@@ -113,15 +117,34 @@ func (e *DataClockConsensusEngine) runFramePruning() {
 				panic(err)
 			}
 
-			if head.FrameNumber < uint64(e.config.Engine.MaxFrames)+1 ||
+			if head.FrameNumber <= maxFrames ||
 				head.FrameNumber <= application.PROOF_FRAME_SENIORITY_REPAIR+1 {
 				continue
 			}
 
-			if err := e.pruneFrames(
-				head.FrameNumber - uint64(e.config.Engine.MaxFrames),
-			); err != nil {
-				e.logger.Error("could not prune", zap.Error(err))
+			to := head.FrameNumber - maxFrames
+			for i := from; i < to; i += batchSize {
+				start, stop := i, min(i+batchSize, to)
+				if err := e.clockStore.DeleteDataClockFrameRange(e.filter, start, stop); err != nil {
+					e.logger.Error(
+						"failed to prune frames",
+						zap.Error(err),
+						zap.Uint64("from", start),
+						zap.Uint64("to", stop),
+					)
+					continue outer
+				}
+				e.logger.Info(
+					"pruned frames",
+					zap.Uint64("from", start),
+					zap.Uint64("to", stop),
+				)
+				select {
+				case <-e.ctx.Done():
+					return
+				default:
+				}
+				from = stop
 			}
 		}
 	}

node/consensus/master/peer_messaging.go

@@ -69,6 +69,7 @@ func (e *MasterClockConsensusEngine) Sync(
 		if err != nil {
 			return errors.Wrap(err, "sync")
 		}
+		defer iter.Close()
 
 		response := []*protobufs.ClockFrame{}
 
@@ -81,10 +82,6 @@ func (e *MasterClockConsensusEngine) Sync(
 			response = append(response, frame)
 		}
 
-		if err = iter.Close(); err != nil {
-			return errors.Wrap(err, "sync")
-		}
-
 		if len(response) == 0 {
 			return nil
 		}

node/main.go

@@ -144,6 +144,11 @@ var (
 		true,
 		"when enabled, frame execution validation is skipped",
 	)
+	compactDB = flag.Bool(
+		"compact-db",
+		false,
+		"compacts the database and exits",
+	)
 )
 
 func signatureCheckDefault() bool {
@@ -330,6 +335,17 @@ func main() {
 		panic(err)
 	}
 
+	if *compactDB && *core == 0 {
+		db := store.NewPebbleDB(nodeConfig.DB)
+		if err := db.CompactAll(); err != nil {
+			panic(err)
+		}
+		if err := db.Close(); err != nil {
+			panic(err)
+		}
+		return
+	}
+
 	if *network != 0 {
 		if nodeConfig.P2P.BootstrapPeers[0] == config.BootstrapPeers[0] {
 			fmt.Println(

node/rpc/node_rpc_server.go

@@ -97,21 +97,17 @@ func (r *RPCServer) GetFrames(
 		if err != nil {
 			return nil, errors.Wrap(err, "get frames")
 		}
+		defer iter.Close()
 
 		frames := []*protobufs.ClockFrame{}
 		for iter.First(); iter.Valid(); iter.Next() {
 			frame, err := iter.Value()
 			if err != nil {
-				iter.Close()
 				return nil, errors.Wrap(err, "get frames")
 			}
 			frames = append(frames, frame)
 		}
 
-		if err := iter.Close(); err != nil {
-			return nil, errors.Wrap(err, "get frames")
-		}
-
 		return &protobufs.FramesResponse{
 			TruncatedClockFrames: frames,
 		}, nil
@@ -124,21 +120,17 @@ func (r *RPCServer) GetFrames(
 		if err != nil {
 			return nil, errors.Wrap(err, "get frame info")
 		}
+		defer iter.Close()
 
 		frames := []*protobufs.ClockFrame{}
 		for iter.First(); iter.Valid(); iter.Next() {
 			frame, err := iter.TruncatedValue()
 			if err != nil {
-				iter.Close()
 				return nil, errors.Wrap(err, "get frames")
 			}
 			frames = append(frames, frame)
 		}
 
-		if err := iter.Close(); err != nil {
-			return nil, errors.Wrap(err, "get frames")
-		}
-
 		return &protobufs.FramesResponse{
 			TruncatedClockFrames: frames,
 		}, nil

node/store/clock.go

@@ -216,13 +216,13 @@ func (p *PebbleClockIterator) TruncatedValue() (
 		if err != nil {
 			return nil, errors.Wrap(err, "get truncated clock frame iterator value")
 		}
+		defer frameCloser.Close()
 		if err := proto.Unmarshal(frameValue, frame); err != nil {
 			return nil, errors.Wrap(
 				errors.Wrap(err, ErrInvalidData.Error()),
 				"get truncated clock frame iterator value",
 			)
 		}
-		frameCloser.Close()
 	} else {
 		if err := proto.Unmarshal(value, frame); err != nil {
 			return nil, errors.Wrap(
@@ -252,13 +252,13 @@ func (p *PebbleClockIterator) Value() (*protobufs.ClockFrame, error) {
 		if err != nil {
 			return nil, errors.Wrap(err, "get clock frame iterator value")
 		}
+		defer frameCloser.Close()
 		if err := proto.Unmarshal(frameValue, frame); err != nil {
 			return nil, errors.Wrap(
 				errors.Wrap(err, ErrInvalidData.Error()),
 				"get clock frame iterator value",
 			)
 		}
-		defer frameCloser.Close()
 	} else {
 		if err := proto.Unmarshal(value, frame); err != nil {
 			return nil, errors.Wrap(
@@ -469,8 +469,8 @@ func (p *PebbleClockStore) GetEarliestMasterClockFrame(
 
 		return nil, errors.Wrap(err, "get earliest master clock frame")
 	}
-
 	defer closer.Close()
+
 	frameNumber := binary.BigEndian.Uint64(idxValue)
 	frame, err := p.GetMasterClockFrame(filter, frameNumber)
 	if err != nil {
@@ -492,8 +492,8 @@ func (p *PebbleClockStore) GetLatestMasterClockFrame(
 
 		return nil, errors.Wrap(err, "get latest master clock frame")
 	}
-
 	defer closer.Close()
+
 	frameNumber := binary.BigEndian.Uint64(idxValue)
 	frame, err := p.GetMasterClockFrame(filter, frameNumber)
 	if err != nil {
@@ -516,11 +516,11 @@ func (p *PebbleClockStore) GetMasterClockFrame(
 
 		return nil, errors.Wrap(err, "get master clock frame")
 	}
+	defer closer.Close()
 
 	copied := make([]byte, len(value))
 	copy(copied[:], value[:])
 
-	defer closer.Close()
 	frame := &protobufs.ClockFrame{}
 	frame.FrameNumber = frameNumber
 	frame.Filter = filter
@@ -595,10 +595,8 @@ func (p *PebbleClockStore) PutMasterClockFrame(
 		); err != nil {
 			return errors.Wrap(err, "put master clock frame")
 		}
-	}
-
-	if err == nil && closer != nil {
-		closer.Close()
+	} else {
+		_ = closer.Close()
 	}
 
 	if err = txn.Set(
@@ -625,6 +623,7 @@ func (p *PebbleClockStore) GetDataClockFrame(
 
 		return nil, nil, errors.Wrap(err, "get data clock frame")
 	}
+	defer closer.Close()
 
 	frame := &protobufs.ClockFrame{}
 	genesisFramePreIndex := false
@@ -641,14 +640,13 @@ func (p *PebbleClockStore) GetDataClockFrame(
 
 			return nil, nil, errors.Wrap(err, "get data clock frame")
 		}
+		defer frameCloser.Close()
 		if err := proto.Unmarshal(frameValue, frame); err != nil {
 			return nil, nil, errors.Wrap(
 				errors.Wrap(err, ErrInvalidData.Error()),
 				"get data clock frame",
 			)
 		}
-		closer.Close()
-		defer frameCloser.Close()
 	} else {
 		genesisFramePreIndex = frameNumber == 0
 		if err := proto.Unmarshal(value, frame); err != nil {
@@ -657,7 +655,6 @@ func (p *PebbleClockStore) GetDataClockFrame(
 				"get data clock frame",
 			)
 		}
-		defer closer.Close()
 	}
 
 	if !truncate {
@@ -674,15 +671,17 @@ func (p *PebbleClockStore) GetDataClockFrame(
 			proverTrie := &tries.RollingFrecencyCritbitTrie{}
 			trieData, closer, err := p.db.Get(clockProverTrieKey(filter, i, frameNumber))
 			if err != nil {
+				if !errors.Is(err, pebble.ErrNotFound) {
+					return nil, nil, errors.Wrap(err, "get data clock frame")
+				}
 				break
 			}
+			defer closer.Close()
 
 			if err := proverTrie.Deserialize(trieData); err != nil {
-				closer.Close()
 				return nil, nil, errors.Wrap(err, "get latest data clock frame")
 			}
 
-			closer.Close()
 			i++
 			proverTries = append(proverTries, proverTrie)
 		}
@@ -726,7 +725,6 @@ func (p *PebbleClockStore) deleteAggregateProofs(
 	for i := 0; i < len(frame.Input[516:])/74; i++ {
 		commit := frame.Input[516+(i*74) : 516+((i+1)*74)]
 		err := internalDeleteAggregateProof(
-			p.db,
 			txn,
 			frame.AggregateProofs[i],
 			commit,
@@ -757,7 +755,6 @@ func (p *PebbleClockStore) saveAggregateProofs(
 	for i := 0; i < len(frame.Input[516:])/74; i++ {
 		commit := frame.Input[516+(i*74) : 516+((i+1)*74)]
 		err := internalPutAggregateProof(
-			p.db,
 			txn,
 			frame.AggregateProofs[i],
 			commit,
@@ -788,8 +785,8 @@ func (p *PebbleClockStore) GetEarliestDataClockFrame(
 
 		return nil, errors.Wrap(err, "get earliest data clock frame")
 	}
-
 	defer closer.Close()
+
 	frameNumber := binary.BigEndian.Uint64(idxValue)
 	frame, _, err := p.GetDataClockFrame(filter, frameNumber, false)
 	if err != nil {
@@ -811,6 +808,7 @@ func (p *PebbleClockStore) GetLatestDataClockFrame(
 
 		return nil, nil, errors.Wrap(err, "get latest data clock frame")
 	}
+	defer closer.Close()
 
 	frameNumber := binary.BigEndian.Uint64(idxValue)
 	frame, tries, err := p.GetDataClockFrame(filter, frameNumber, false)
@@ -822,8 +820,6 @@ func (p *PebbleClockStore) GetLatestDataClockFrame(
 		return nil, nil, errors.Wrap(err, "get latest data clock frame")
 	}
 
-	closer.Close()
-
 	return frame, tries, nil
 }
 
@@ -843,6 +839,7 @@ func (p *PebbleClockStore) GetStagedDataClockFrame(
 		}
 		return nil, errors.Wrap(err, "get parent data clock frame")
 	}
+	defer closer.Close()
 
 	parent := &protobufs.ClockFrame{}
 	if err := proto.Unmarshal(data, parent); err != nil {
@@ -858,10 +855,6 @@ func (p *PebbleClockStore) GetStagedDataClockFrame(
 		}
 	}
 
-	if closer != nil {
-		closer.Close()
-	}
-
 	return parent, nil
 }
 
@@ -879,9 +872,9 @@ func (p *PebbleClockStore) GetStagedDataClockFramesForFrameNumber(
 		}
 		return nil, errors.Wrap(err, "get staged data clock frames")
 	}
+	defer iter.Close()
 
 	frames := []*protobufs.ClockFrame{}
-
 	for iter.First(); iter.Valid(); iter.Next() {
 		data := iter.Value()
 		frame := &protobufs.ClockFrame{}
@@ -899,8 +892,6 @@ func (p *PebbleClockStore) GetStagedDataClockFramesForFrameNumber(
 		frames = append(frames, frame)
 	}
 
-	iter.Close()
-
 	return frames, nil
 }
 
@@ -992,10 +983,8 @@ func (p *PebbleClockStore) CommitDataClockFrame(
 		); err != nil {
 			return errors.Wrap(err, "commit data clock frame")
 		}
-	}
-
-	if err == nil && closer != nil {
-		closer.Close()
+	} else {
+		_ = closer.Close()
 	}
 
 	if !backfill {
@@ -1058,36 +1047,99 @@ func (p *PebbleClockStore) DeleteDataClockFrameRange(
 	for i := fromFrameNumber; i < toFrameNumber; i++ {
 		frames, err := p.GetStagedDataClockFramesForFrameNumber(filter, i)
 		if err != nil {
-			continue
+			if !errors.Is(err, ErrNotFound) && !errors.Is(err, ErrInvalidData) {
+				_ = txn.Abort()
+				return errors.Wrap(err, "delete data clock frame range")
+			}
+			frames = nil
 		}
 
+	outer:
 		for _, frame := range frames {
-			err = p.deleteAggregateProofs(txn, frame)
-			if err != nil {
-				continue
+			for _, ap := range frame.AggregateProofs {
+				for _, inc := range ap.InclusionCommitments {
+					// The commitments collide for very small frames, and as such we have to detect them early
+					// and avoid deleting them. Common cases for such collisions are prover announcement messages
+					// which do not contain the frame number, so their binary contents are equivalent between
+					// multiple frames.
+					if len(inc.Data) < 2048 {
+						continue outer
+					}
+					if inc.TypeUrl == protobufs.IntrinsicExecutionOutputType {
+						o := &protobufs.IntrinsicExecutionOutput{}
+						if err := proto.Unmarshal(inc.Data, o); err != nil {
+							_ = txn.Abort()
+							return errors.Wrap(err, "delete data clock frame range")
+						}
+						// The commitments collide for empty frames, and as such we have to detect them early
+						// and avoid deleting them.
+						if len(o.Output) == 0 || len(o.Proof) == 0 {
+							continue outer
+						}
+					}
+				}
+			}
+			if err := p.deleteAggregateProofs(txn, frame); err != nil {
+				if !errors.Is(err, pebble.ErrNotFound) {
+					_ = txn.Abort()
+					return errors.Wrap(err, "delete data clock frame range")
+				}
 			}
 		}
 
-		err = txn.DeleteRange(
+		if err := txn.DeleteRange(
 			clockDataParentIndexKey(filter, i, bytes.Repeat([]byte{0x00}, 32)),
 			clockDataParentIndexKey(filter, i, bytes.Repeat([]byte{0xff}, 32)),
-		)
-		if err != nil {
-			continue
+		); err != nil {
+			if !errors.Is(err, pebble.ErrNotFound) {
+				_ = txn.Abort()
+				return errors.Wrap(err, "delete data clock frame range")
+			}
 		}
 
-		err = txn.Delete(clockDataFrameKey(filter, i))
-		if err != nil {
-			continue
+		if err := txn.Delete(clockDataFrameKey(filter, i)); err != nil {
+			if !errors.Is(err, pebble.ErrNotFound) {
+				_ = txn.Abort()
+				return errors.Wrap(err, "delete data clock frame range")
+			}
+		}
+
+		// The prover trie keys are not stored continuously with respect
+		// to the same frame number. As such, we need to manually iterate
+		// and discover such keys.
+		for t := uint16(0); t <= 0xffff; t++ {
+			_, closer, err := p.db.Get(clockProverTrieKey(filter, t, i))
+			if err != nil {
+				if !errors.Is(err, pebble.ErrNotFound) {
+					_ = txn.Abort()
+					return errors.Wrap(err, "delete data clock frame range")
+				} else {
+					break
+				}
+			}
+			_ = closer.Close()
+			if err := txn.Delete(clockProverTrieKey(filter, t, i)); err != nil {
+				_ = txn.Abort()
+				return errors.Wrap(err, "delete data clock frame range")
+			}
+		}
+
+		if err := txn.DeleteRange(
+			clockDataTotalDistanceKey(filter, i, bytes.Repeat([]byte{0x00}, 32)),
+			clockDataTotalDistanceKey(filter, i, bytes.Repeat([]byte{0xff}, 32)),
+		); err != nil {
+			if !errors.Is(err, pebble.ErrNotFound) {
+				_ = txn.Abort()
+				return errors.Wrap(err, "delete data clock frame range")
+			}
 		}
 	}
 
-	if err = txn.Commit(); err != nil {
-		txn.Abort()
+	if err := txn.Commit(); err != nil {
 		return errors.Wrap(err, "delete data clock frame range")
 	}
 
-	return errors.Wrap(err, "delete data clock frame range")
+	return nil
 }
 
 func (p *PebbleClockStore) ResetMasterClockFrames(filter []byte) error {
@@ -1138,7 +1190,7 @@ func (p *PebbleClockStore) Compact(
 		if bytes.Compare(version, config.GetVersion()) < 0 {
 			cleared = false
 		}
-		closer.Close()
+		defer closer.Close()
 	}
 
 	if !cleared {
@@ -1190,11 +1242,10 @@ func (p *PebbleClockStore) Compact(
 
 			return errors.Wrap(err, "compact")
 		}
+		defer closer.Close()
 
 		last := binary.BigEndian.Uint64(idxValue)
 
-		closer.Close()
-
 		for frameNumber := uint64(1); frameNumber <= last; frameNumber++ {
 			value, closer, err := p.db.Get(clockDataFrameKey(dataFilter, frameNumber))
 			if err != nil {
@@ -1204,6 +1255,7 @@ func (p *PebbleClockStore) Compact(
 
 				return errors.Wrap(err, "compact")
 			}
+			defer closer.Close()
 
 			frame := &protobufs.ClockFrame{}
 
@@ -1212,6 +1264,7 @@ func (p *PebbleClockStore) Compact(
 				if err != nil {
 					return errors.Wrap(err, "compact")
 				}
+				defer frameCloser.Close()
 				if err := proto.Unmarshal(frameValue, frame); err != nil {
 					return errors.Wrap(err, "compact")
 				}
@@ -1226,9 +1279,6 @@ func (p *PebbleClockStore) Compact(
 						make([]byte, 32),
 					)),
 				)
-
-				closer.Close()
-				frameCloser.Close()
 			} else {
 				if err := proto.Unmarshal(value, frame); err != nil {
 					return errors.Wrap(err, "compact")
@@ -1255,14 +1305,15 @@ func (p *PebbleClockStore) Compact(
 						make([]byte, 32),
 					)),
 				)
+				if err != nil {
+					return errors.Wrap(err, "compact")
+				}
 
 				parents = append(parents,
 					clockDataParentIndexKey(dataFilter, frameNumber, selector.FillBytes(
 						make([]byte, 32),
 					)),
 				)
-
-				closer.Close()
 			}
 
 			for i := 0; i < len(frame.Input[516:])/74; i++ {
@@ -1523,10 +1574,8 @@ func (p *PebbleClockStore) GetTotalDistance(
 
 		return nil, errors.Wrap(err, "get total distance")
 	}
-
 	defer closer.Close()
 	dist := new(big.Int).SetBytes(value)
-
 	return dist, nil
 }
 
@@ -1564,7 +1613,6 @@ func (p *PebbleClockStore) GetPeerSeniorityMap(filter []byte) (
 	if err = dec.Decode(&seniorityMap); err != nil {
 		return nil, errors.Wrap(err, "get peer seniority map")
 	}
-
 	return seniorityMap, nil
 }
 
@@ -1612,9 +1660,12 @@ func (p *PebbleClockStore) SetProverTriesForFrame(
 			clockProverTrieKey(frame.Filter, uint16(start), frame.FrameNumber),
 		)
 		if err != nil {
+			if !errors.Is(err, pebble.ErrNotFound) {
+				return errors.Wrap(err, "set prover tries for frame")
+			}
 			break
 		}
-		closer.Close()
+		_ = closer.Close()
 
 		if err = p.db.Delete(
 			clockProverTrieKey(frame.Filter, uint16(start), frame.FrameNumber),

node/store/coin.go

@@ -136,8 +136,8 @@ func (p *PebbleCoinStore) GetCoinsForOwner(
 		err = errors.Wrap(err, "get coins for owner")
 		return nil, nil, nil, err
 	}
-
 	defer iter.Close()
+
 	frameNumbers := []uint64{}
 	addresses := [][]byte{}
 	coins := []*protobufs.Coin{}
@@ -176,8 +176,8 @@ func (p *PebbleCoinStore) GetPreCoinProofsForOwner(owner []byte) (
 		err = errors.Wrap(err, "get pre coin proofs for owner")
 		return nil, nil, err
 	}
-
 	defer iter.Close()
+
 	frameNumbers := []uint64{}
 	proofs := []*protobufs.PreCoinProof{}
 	for iter.First(); iter.Valid(); iter.Next() {
@@ -215,7 +215,6 @@ func (p *PebbleCoinStore) GetCoinByAddress(txn Transaction, address []byte) (
 		err = errors.Wrap(err, "get coin by address")
 		return nil, err
 	}
-
 	defer closer.Close()
 
 	coin := &protobufs.Coin{}
@@ -240,7 +239,6 @@ func (p *PebbleCoinStore) GetPreCoinProofByAddress(address []byte) (
 		err = errors.Wrap(err, "get pre coin proof by address")
 		return nil, err
 	}
-
 	defer closer.Close()
 
 	proof := &protobufs.PreCoinProof{}
@@ -386,9 +384,9 @@ func (p *PebbleCoinStore) GetLatestFrameProcessed() (uint64, error) {
 
 		return 0, errors.Wrap(err, "get latest frame processed")
 	}
+	defer closer.Close()
 
 	frameNumber := binary.BigEndian.Uint64(v)
-	closer.Close()
 
 	return frameNumber, nil
 }
@@ -524,13 +522,12 @@ func (p *PebbleCoinStore) Migrate(filter []byte, genesisSeedHex string) error {
 		}
 		return p.internalMigrate(filter, seed)
 	}
+	defer closer.Close()
 
 	if !bytes.Equal(compare, seed) {
 		return p.internalMigrate(filter, seed)
 	}
 
-	closer.Close()
-
 	status, closer, err := p.db.Get(migrationKey())
 	if err != nil {
 		if !errors.Is(err, pebble.ErrNotFound) {

node/store/data_proof.go

@@ -139,8 +139,8 @@ func internalGetAggregateProof(
 
 		return nil, errors.Wrap(err, "get aggregate proof")
 	}
-
 	defer closer.Close()
+
 	copied := make([]byte, len(value[8:]))
 	limit := binary.BigEndian.Uint64(value[0:8])
 	copy(copied, value[8:])
@@ -159,6 +159,7 @@ func internalGetAggregateProof(
 	if err != nil {
 		return nil, errors.Wrap(err, "get aggregate proof")
 	}
+	defer iter.Close()
 
 	i := uint32(0)
 
@@ -199,14 +200,11 @@ func internalGetAggregateProof(
 
 				return nil, errors.Wrap(err, "get aggregate proof")
 			}
+			defer dataCloser.Close()
 
 			segCopy := make([]byte, len(segValue))
 			copy(segCopy, segValue)
 			chunks = append(chunks, segCopy)
-
-			if err = dataCloser.Close(); err != nil {
-				return nil, errors.Wrap(err, "get aggregate proof")
-			}
 		}
 
 		if string(url) == protobufs.IntrinsicExecutionOutputType {
@@ -236,10 +234,6 @@ func internalGetAggregateProof(
 		i++
 	}
 
-	if err = iter.Close(); err != nil {
-		return nil, errors.Wrap(err, "get aggregate proof")
-	}
-
 	return aggregate, nil
 }
 
@@ -263,8 +257,8 @@ func internalListAggregateProofKeys(
 
 		return nil, nil, nil, errors.Wrap(err, "list aggregate proof")
 	}
-
 	defer closer.Close()
+
 	copied := make([]byte, len(value[8:]))
 	limit := binary.BigEndian.Uint64(value[0:8])
 	copy(copied, value[8:])
@@ -278,6 +272,7 @@ func internalListAggregateProofKeys(
 
 		return nil, nil, nil, errors.Wrap(err, "list aggregate proof")
 	}
+	defer iter.Close()
 
 	i := uint32(0)
 	commits = append(commits, dataProofInclusionKey(filter, commitment, 0))
@@ -305,10 +300,6 @@ func internalListAggregateProofKeys(
 		i++
 	}
 
-	if err = iter.Close(); err != nil {
-		return nil, nil, nil, errors.Wrap(err, "list aggregate proof")
-	}
-
 	return proofs, commits, data, nil
 }
 
@@ -326,17 +317,10 @@ func (p *PebbleDataProofStore) GetAggregateProof(
 }
 
 func internalDeleteAggregateProof(
-	db KVDB,
 	txn Transaction,
 	aggregateProof *protobufs.InclusionAggregateProof,
 	commitment []byte,
 ) error {
-	buf := binary.BigEndian.AppendUint64(
-		nil,
-		uint64(len(aggregateProof.InclusionCommitments)),
-	)
-	buf = append(buf, aggregateProof.Proof...)
-
 	for i, inc := range aggregateProof.InclusionCommitments {
 		var segments [][]byte
 		if inc.TypeUrl == protobufs.IntrinsicExecutionOutputType {
@@ -378,7 +362,6 @@ func internalDeleteAggregateProof(
 }
 
 func internalPutAggregateProof(
-	db KVDB,
 	txn Transaction,
 	aggregateProof *protobufs.InclusionAggregateProof,
 	commitment []byte,
@@ -447,7 +430,6 @@ func (p *PebbleDataProofStore) PutAggregateProof(
 	commitment []byte,
 ) error {
 	return internalPutAggregateProof(
-		p.db,
 		txn,
 		aggregateProof,
 		commitment,
@@ -467,8 +449,8 @@ func (p *PebbleDataProofStore) GetDataTimeProof(
 		err = errors.Wrap(err, "get data time proof")
 		return
 	}
-
 	defer closer.Close()
+
 	if len(data) < 24 {
 		err = ErrInvalidData
 		return
@@ -513,13 +495,10 @@ func (p *PebbleDataProofStore) GetTotalReward(
 
 		return nil, errors.Wrap(err, "get total difficulty sum")
 	}
+	defer closer.Close()
 
 	if len(prev) != 0 {
 		reward.SetBytes(prev[4:])
-
-		if err = closer.Close(); err != nil {
-			return nil, errors.Wrap(err, "get total difficulty sum")
-		}
 	}
 
 	return reward, nil
@@ -556,15 +535,12 @@ func (p *PebbleDataProofStore) PutDataTimeProof(
 	if err != nil && (!errors.Is(err, pebble.ErrNotFound) || increment != 0) {
 		return errors.Wrap(err, "put data time proof")
 	}
+	defer closer.Close()
 
 	if len(prev) != 0 {
 		priorSum.SetBytes(prev[4:])
 		prevIncrement := binary.BigEndian.Uint32(prev[:4])
 
-		if err = closer.Close(); err != nil {
-			return errors.Wrap(err, "put data time proof")
-		}
-
 		if prevIncrement != increment-1 {
 			return errors.Wrap(errors.New("invalid increment"), "put data time proof")
 		}
@@ -609,15 +585,13 @@ func (p *PebbleDataProofStore) GetLatestDataTimeProof(peerId []byte) (
 
 		return 0, 0, nil, errors.Wrap(err, "get latest data time proof")
 	}
+	defer closer.Close()
 
 	if len(prev) < 4 {
 		return 0, 0, nil, ErrInvalidData
 	}
 
 	increment = binary.BigEndian.Uint32(prev[:4])
-	if err = closer.Close(); err != nil {
-		return 0, 0, nil, errors.Wrap(err, "get latest data time proof")
-	}
 
 	_, parallelism, _, output, err = p.GetDataTimeProof(peerId, increment)
 

node/store/inmem.go

@@ -281,15 +281,13 @@ func (t *InMemKVDBTransaction) DeleteRange(
 	if err != nil {
 		return err
 	}
+	defer iter.Close()
 
 	for iter.First(); iter.Valid(); iter.Next() {
 		t.changes = append(t.changes, InMemKVDBOperation{
 			op:  DeleteOperation,
 			key: iter.Key(),
 		})
-		if err != nil {
-			return err
-		}
 	}
 
 	return nil
@@ -416,6 +414,7 @@ func (d *InMemKVDB) DeleteRange(start, end []byte) error {
 	if err != nil {
 		return err
 	}
+	defer iter.Close()
 
 	for iter.First(); iter.Valid(); iter.Next() {
 		err = d.Delete(iter.Key())

node/store/key.go

@@ -270,6 +270,8 @@ func (p *PebbleKeyStore) IncludeProvingKey(
 	staged, closer, err := p.db.Get(stagedProvingKeyKey(provingKey.PublicKey()))
 	if err != nil && !errors.Is(err, ErrNotFound) {
 		return errors.Wrap(err, "include proving key")
+	} else if err == nil {
+		defer closer.Close()
 	}
 
 	if staged != nil {
@@ -279,9 +281,6 @@ func (p *PebbleKeyStore) IncludeProvingKey(
 			return errors.Wrap(err, "include proving key")
 		}
 	}
-	if err := closer.Close(); err != nil {
-		return errors.Wrap(err, "include proving key")
-	}
 
 	return nil
 }
@@ -297,16 +296,13 @@ func (p *PebbleKeyStore) GetStagedProvingKey(
 
 		return nil, errors.Wrap(err, "get staged proving key")
 	}
+	defer closer.Close()
 
 	stagedKey := &protobufs.ProvingKeyAnnouncement{}
 	if err = proto.Unmarshal(data, stagedKey); err != nil {
 		return nil, errors.Wrap(err, "get staged proving key")
 	}
 
-	if err := closer.Close(); err != nil {
-		return nil, errors.Wrap(err, "get staged proving key")
-	}
-
 	return stagedKey, nil
 }
 
@@ -322,12 +318,9 @@ func (p *PebbleKeyStore) GetLatestKeyBundle(
 
 		return nil, errors.Wrap(err, "get latest key bundle")
 	}
+	defer closer.Close()
 	frameNumber := binary.BigEndian.Uint64(value)
 
-	if err := closer.Close(); err != nil {
-		return nil, errors.Wrap(err, "get latest key bundle")
-	}
-
 	value, closer, err = p.db.Get(keyBundleKey(provingKey, frameNumber))
 	if err != nil {
 		if errors.Is(err, pebble.ErrNotFound) {
@@ -336,7 +329,6 @@ func (p *PebbleKeyStore) GetLatestKeyBundle(
 
 		return nil, errors.Wrap(err, "get latest key bundle")
 	}
-
 	defer closer.Close()
 
 	announcement := &protobufs.InclusionCommitment{}
@@ -440,10 +432,8 @@ func (p *PebbleKeyStore) PutKeyBundle(
 		); err != nil {
 			return errors.Wrap(err, "put key bundle")
 		}
-	}
-
-	if err == nil && closer != nil {
-		closer.Close()
+	} else {
+		_ = closer.Close()
 	}
 
 	if err = txn.Set(

node/store/peerstore.go

@@ -74,10 +74,10 @@ func (d *PeerstoreDatastore) Get(
 		}
 		return nil, err
 	}
+	defer closer.Close()
 
 	out := make([]byte, len(val))
 	copy(out[:], val[:])
-	closer.Close()
 
 	return val, nil
 }
@@ -226,10 +226,10 @@ func (t *transaction) Get(
 		}
 		return nil, errors.Wrap(err, "get")
 	}
+	defer closer.Close()
 
 	out := make([]byte, len(b))
 	copy(out[:], b[:])
-	closer.Close()
 
 	return b, nil
 }