prevent "NewPartial" and "UndoPartial" steps from getting to user through cursor in dataOutput, fix partial blocks output in v4

Author: sduchesneau

pipeline/pipeline.go

@@ -919,25 +919,51 @@ func (p *Pipeline) cleanUpModuleExecutors(ctx context.Context, logger *zap.Logge
 	return nil
 }
 
+// normalizedOpaqueCursor returns a opaque cursor string without some
+// esoteric steps like 'NewPartial' and 'UndoPartial' which may not be supported by some clients
+func normalizedOpaqueCursor(cursor bstream.Cursor) string {
+	switch cursor.Step {
+	case bstream.StepNewPartial:
+		cursor.Step = bstream.StepNew
+	case bstream.StepUndoPartial:
+		cursor.Step = bstream.StepUndo
+	}
+	return cursor.ToOpaque()
+}
+
 func returnPartialDataOutput(
 	clock *pbsubstreams.Clock,
 	cursor *bstream.Cursor,
 	mapModuleOutput *pbsubstreamsrpc.MapModuleOutput,
 	respFunc substreams.ResponseFunc,
 	partialIdx uint32,
 	lastPartial bool,
+	supportBuffering bool,
 ) error {
 
 	out := &pbsubstreamsrpc.BlockScopedData{
 		Clock:            clock,
-		Cursor:           cursor.ToOpaque(),
+		Cursor:           normalizedOpaqueCursor(*cursor),
 		FinalBlockHeight: cursor.LIB.Num(),
 		Output:           mapModuleOutput,
 		IsPartial:        true,
 		PartialIndex:     &partialIdx,
 		IsLastPartial:    &lastPartial,
 	}
 
+	if supportBuffering { //v4 support
+		bsd := &pbsubstreamsrpcv4.BlockScopedDatas{
+			Items: []*pbsubstreamsrpc.BlockScopedData{
+				out,
+			},
+		}
+
+		if err := respFunc(substreams.NewBlockScopedDatasResponse(bsd)); err != nil {
+			return fmt.Errorf("calling return func: %w", err)
+		}
+		return nil
+	}
+
 	if err := respFunc(substreams.NewBlockScopedDataResponse(out)); err != nil {
 		return fmt.Errorf("calling response func: %w", err)
 	}
@@ -966,7 +992,7 @@ func returnModuleDataOutputs(
 		Output:            mapModuleOutput,
 		DebugMapOutputs:   extraMapModuleOutputs,
 		DebugStoreOutputs: extraStoreModuleOutputs,
-		Cursor:            cursor.ToOpaque(),
+		Cursor:            normalizedOpaqueCursor(*cursor),
 		FinalBlockHeight:  cursor.LIB.Num(),
 	}
 

pipeline/process_block.go

@@ -272,7 +272,7 @@ func (p *Pipeline) handleUndo(clock *pbsubstreams.Clock, cursor *bstream.Cursor,
 			Message: &pbsubstreamsrpc.Response_BlockUndoSignal{
 				BlockUndoSignal: &pbsubstreamsrpc.BlockUndoSignal{
 					LastValidBlock:  targetClock,
-					LastValidCursor: targetCursor.ToOpaque(),
+					LastValidCursor: normalizedOpaqueCursor(*targetCursor),
 				},
 			},
 		})
@@ -399,7 +399,7 @@ func (p *Pipeline) handleStepPartial(ctx context.Context, clock *pbsubstreams.Cl
 
 	mapModuleOutput := normalizeModuleOutput(p.mapModuleOutput, reqDetails.OutputModule)
 
-	if err = returnPartialDataOutput(clock, cursor, mapModuleOutput, p.respFunc, uint32(idx), isLast); err != nil {
+	if err = returnPartialDataOutput(clock, cursor, mapModuleOutput, p.respFunc, uint32(idx), isLast, p.supportBuffering); err != nil {
 		return fmt.Errorf("failed to return module data output: %w", err)
 	}
 

pipeline/resolve.go

@@ -222,7 +222,7 @@ func resolveStartBlockNum(ctx context.Context, req *pbsubstreamsrpc.Request, res
 				Number: cursor.Block.Num(),
 				Id:     cursor.Block.ID(),
 			},
-			LastValidCursor: cursor.ToOpaque(),
+			LastValidCursor: normalizedOpaqueCursor(*cursor),
 		}
 	}
 
@@ -255,7 +255,7 @@ func resolveStartBlockNum(ctx context.Context, req *pbsubstreamsrpc.Request, res
 
 		undoSignal = &pbsubstreamsrpc.BlockUndoSignal{
 			LastValidBlock:  blockRefToPB(reorgJunctionBlock),
-			LastValidCursor: resolvedCursor.ToOpaque(),
+			LastValidCursor: normalizedOpaqueCursor(*resolvedCursor),
 		}
 	}
 
@@ -267,7 +267,7 @@ func resolveStartBlockNum(ctx context.Context, req *pbsubstreamsrpc.Request, res
 		resolvedStartBlockNum = resolvedCursor.Block.Num()
 	}
 
-	return resolvedStartBlockNum, resolvedCursor.ToOpaque(), undoSignal, nil
+	return resolvedStartBlockNum, normalizedOpaqueCursor(*resolvedCursor), undoSignal, nil
 }
 
 type CursorResolver func(context.Context, *bstream.Cursor) (reorgJunctionBlock, currentHead bstream.BlockRef, err error)