misc(store): various fixes and chores (#330)

* Fixes leftover bug from #326 
* refactors store wiping case to avoid silent failure for setTail
* fixes two bugs in deleteRange
  * context shadowing bug causing context to cancel prematurely
  * closed of a closed channel panic when multiple workers error
*  error wrappings
* more logging improvements

Author: Wondertan

store/store.go

@@ -149,6 +149,7 @@ func (s *Store[H]) Stop(ctx context.Context) error {
 	// signal to prevent further writes to Store
 	select {
 	case s.writes <- nil:
+		s.cancel()
 	case <-ctx.Done():
 		return ctx.Err()
 	}
@@ -362,9 +363,6 @@ func (s *Store[H]) HasAt(ctx context.Context, height uint64) bool {
 
 func (s *Store[H]) setTail(ctx context.Context, write datastore.Write, to uint64) error {
 	newTail, err := s.getByHeight(ctx, to)
-	if errors.Is(err, header.ErrNotFound) {
-		return nil
-	}
 	if err != nil {
 		return fmt.Errorf("getting tail: %w", err)
 	}

store/store_delete.go

@@ -70,16 +70,18 @@ func (s *Store[H]) DeleteTo(ctx context.Context, to uint64) error {
 		return fmt.Errorf("header/store: delete to %d below current tail(%d)", to, tail.Height())
 	}
 
-	if err := s.deleteRange(ctx, tail.Height(), to); err != nil {
-		return fmt.Errorf("header/store: delete to height %d: %w", to, err)
-	}
-
-	if head.Height()+1 == to {
+	err = s.deleteRange(ctx, tail.Height(), to)
+	if errors.Is(err, header.ErrNotFound) && head.Height()+1 == to {
 		// this is the case where we have deleted all the headers
 		// wipe the store
 		if err := s.wipe(ctx); err != nil {
 			return fmt.Errorf("header/store: wipe: %w", err)
 		}
+
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("header/store: delete to height %d: %w", to, err)
 	}
 
 	return nil
@@ -117,28 +119,29 @@ func (s *Store[H]) deleteRange(ctx context.Context, from, to uint64) (err error)
 				)
 			}
 		} else if to-from > 1 {
-			log.Debugw("deleted headers", "from_height", from, "to_height", to, "took", time.Since(startTime))
+			log.Debugw("deleted headers", "from_height", from, "to_height", to, "took(s)", time.Since(startTime).Seconds())
 		}
 
 		if derr := s.setTail(ctx, s.ds, height); derr != nil {
 			err = errors.Join(err, fmt.Errorf("setting tail to %d: %w", height, derr))
 		}
 	}()
 
+	deleteCtx := ctx
 	if deadline, ok := ctx.Deadline(); ok {
 		// allocate 95% of caller's set deadline for deletion
 		// and give leftover to save progress
 		// this prevents store's state corruption from partial deletion
 		sub := deadline.Sub(startTime) / 100 * 95
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithDeadlineCause(ctx, startTime.Add(sub), errDeleteTimeout)
+		deleteCtx, cancel = context.WithDeadlineCause(ctx, startTime.Add(sub), errDeleteTimeout)
 		defer cancel()
 	}
 
 	if to-from < deleteRangeParallelThreshold {
-		height, err = s.deleteSequential(ctx, from, to)
+		height, err = s.deleteSequential(deleteCtx, from, to)
 	} else {
-		height, err = s.deleteParallel(ctx, from, to)
+		height, err = s.deleteParallel(deleteCtx, from, to)
 	}
 
 	return err
@@ -159,7 +162,7 @@ func (s *Store[H]) deleteSingle(
 
 	hash, err := s.heightIndex.HashByHeight(ctx, height, false)
 	if errors.Is(err, datastore.ErrNotFound) {
-		log.Warnw("attempt to delete header that's not found", "height", height)
+		log.Debugw("attempt to delete header that's not found", "height", height)
 		return nil
 	}
 	if err != nil {
@@ -219,19 +222,21 @@ func (s *Store[H]) deleteSequential(
 // deleteParallel deletes [from:to) header range from the store in parallel
 // and returns the highest unprocessed height: 'to' in success case or the failed height in error case.
 func (s *Store[H]) deleteParallel(ctx context.Context, from, to uint64) (uint64, error) {
-	log.Debugw("starting delete range parallel", "from_height", from, "to_height", to)
-
+	now := time.Now()
 	s.onDeleteMu.Lock()
 	onDelete := slices.Clone(s.onDelete)
 	s.onDeleteMu.Unlock()
-
 	// workerNum defines how many parallel delete workers to run
 	// Scales of number of CPUs configured for the process.
 	// Usually, it's recommended to have 2-4 multiplier for the number of CPUs for
 	// IO operations. Three was picked empirically to be a sweet spot that doesn't
 	// require too much RAM, yet shows good performance.
 	workerNum := runtime.GOMAXPROCS(-1) * 3
 
+	log.Infow(
+		"deleting range parallel", "from_height", from, "to_height", to, "worker_num", workerNum,
+	)
+
 	type result struct {
 		height uint64
 		err    error
@@ -245,7 +250,11 @@ func (s *Store[H]) deleteParallel(ctx context.Context, from, to uint64) (uint64,
 		defer func() {
 			results[worker] = last
 			if last.err != nil {
-				close(errCh)
+				select {
+				case <-errCh:
+				default:
+					close(errCh)
+				}
 			}
 		}()
 
@@ -277,11 +286,11 @@ func (s *Store[H]) deleteParallel(ctx context.Context, from, to uint64) (uint64,
 		}(i)
 	}
 
-	for i, height := 0, from; height < to; height++ {
+	for i, height := uint64(0), from; height < to; height++ {
 		select {
 		case jobCh <- height:
 			i++
-			if uint64(1)%deleteRangeParallelThreshold == 0 {
+			if i%deleteRangeParallelThreshold == 0 {
 				log.Debugf("deleting %dth header height %d", deleteRangeParallelThreshold, height)
 			}
 		case <-errCh:
@@ -312,5 +321,14 @@ func (s *Store[H]) deleteParallel(ctx context.Context, from, to uint64) (uint64,
 
 	// ensures the height after the highest deleted becomes the new tail
 	highest++
+	log.Infow(
+		"deleted range parallel",
+		"from_height",
+		from,
+		"to_height",
+		to,
+		"took(s)",
+		time.Since(now).Seconds(),
+	)
 	return highest, nil
 }

sync/syncer_head.go

@@ -178,7 +178,12 @@ func (s *Syncer[H]) localHead(ctx context.Context) (H, error) {
 		return pendHead, nil
 	}
 	// if pending is empty - get the latest stored/synced head
-	return s.store.Head(ctx)
+	head, err := s.store.Head(ctx)
+	if err != nil {
+		return head, fmt.Errorf("local store head: %w", err)
+	}
+
+	return head, nil
 }
 
 // setLocalHead takes the already validated head and sets it as the new sync target.