fix: potential misalignments in batch sizes

Author: setavenger

internal/database/dbpebble/compute_index.go

@@ -195,12 +195,14 @@ func (s *Store) FinishComputeIndex(height uint32, computeIndexes []ComputeIndex)
 	s.batchSync.Lock()
 	err := attachComputeIndexToBatch(s.dbBatch, computeIndexes)
 	if err != nil {
+		s.batchSync.Unlock()
 		return err
 	}
+	s.batchCounter++
+	shouldCommit := s.batchCounter >= 200
 	s.batchSync.Unlock()
 
-	s.batchCounter++
-	if s.batchCounter >= 200 {
+	if shouldCommit {
 		err = s.commitBatch(false)
 		if err != nil {
 			return err

internal/database/dbpebble/store.go

@@ -39,6 +39,8 @@ func NewStore(db *pebble.DB) *Store {
 }
 
 func (s *Store) BatchSize() int {
+	s.batchSync.Lock()
+	defer s.batchSync.Unlock()
 	return s.batchCounter
 }
 
@@ -58,10 +60,11 @@ func (s *Store) collectAndWrite(block *database.DBBlock) error {
 	if s.dbBatch == nil {
 		s.dbBatch = s.DB.NewBatch()
 	}
+	s.batchCounter++
+	shouldCommit := s.batchCounter > s.batchSize
 	s.batchSync.Unlock()
 
-	s.batchCounter++
-	if s.batchCounter > s.batchSize {
+	if shouldCommit {
 		// rotates batch and commits the old one in the background
 		if err := s.commitBatch(false); err != nil {
 			logging.L.Err(err).Msg("failed to commit batch")
@@ -85,11 +88,15 @@ func (s *Store) attachBlockToBatch(block *database.DBBlock) error {
 }
 
 func (s *Store) FlushBatch(sync bool) error {
-	if s.batchCounter == 0 {
+	s.batchSync.Lock()
+	counter := s.batchCounter
+	s.batchSync.Unlock()
+
+	if counter == 0 {
 		return nil
 	}
 	logging.L.Info().
-		Int("batch_counter", s.batchCounter).
+		Int("batch_counter", counter).
 		Bool("sync", sync).
 		Msg("flushing batch")