clarify error message on cursor resolution

cursor_resolver.go

@@ -3,6 +3,7 @@ package bstream
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
@@ -11,6 +12,8 @@ import (
 	"go.uber.org/zap"
 )
 
+var ErrResolveCursor = errors.New("cannot resolve cursor")
+
 // cursorResolver is a handler that feeds from a source of new+irreversible blocks (filesource)
 // and keeps blocks in a slice until cursor is passed.
 // when it sees the cursor, it sends whatever is needed to bring the consumer back to a "new and irreversible" head
@@ -177,11 +180,11 @@ func (f *cursorResolver) resolve(ctx context.Context) (undoBlocks []*Block, cont
 
 		forkedBlock := oneBlocks[previousID]
 		if forkedBlock == nil {
-			return nil, 0, fmt.Errorf("cannot resolve cursor pointing to block %s: missing link: no one-block-file or merged block found with ID %s", block, previousID)
+			return nil, 0, fmt.Errorf("%w: missing link between blocks %d and %s: no forked-block file found with ID ending with %s.", ErrResolveCursor, lib.Num(), block, previousID)
 		}
 
 		if forkedBlock.Num < lib.Num() {
-			return nil, 0, fmt.Errorf("cannot resolve cursor pointing to block %s: missing link: forked chain goes beyond LIB, looking for ID %s (this should not happens)", block, previousID)
+			return nil, 0, fmt.Errorf("%w: block %s not linkable to canonical chain above final block %d (looking for ID ending with %s)", ErrResolveCursor, block, lib.Num(), previousID)
 		}
 
 		previousID = forkedBlock.PreviousID

filesource.go

@@ -559,7 +559,7 @@ func (s *FileSource) launchSink() {
 				}
 
 				if err := s.handler.ProcessBlock(preBlock.Block, preBlock.Obj); err != nil {
-					s.Shutdown(fmt.Errorf("process block failed: %w", err))
+					s.Shutdown(err)
 					return
 				}
 				if s.highestFileProcessedBlock != nil && preBlock.Num() > s.highestFileProcessedBlock.Num() {

stream/stream.go

@@ -2,6 +2,7 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/streamingfast/bstream"
@@ -92,6 +93,9 @@ func (s *Stream) Run(ctx context.Context) error {
 	source.Run()
 	if err := source.Err(); err != nil {
 		s.logger.Debug("source shutting down", zap.Error(err))
+		if errors.Is(err, bstream.ErrResolveCursor) {
+			return &ErrInvalidArg{message: err.Error()}
+		}
 		return err
 	}
 	return nil

util.go

@@ -95,7 +95,7 @@ func (s *preMergeBlockSource) Run() {
 			s.Shutdown(err)
 		}
 		err = s.handler.ProcessBlock(nativeBlock, nil)
-		s.Shutdown(fmt.Errorf("process block failed: %w", err))
+		s.Shutdown(err)
 	}
 
 	s.Shutdown(nil)