Add reader-node-start-block-timestamp gating across reader-node ingestion paths

cmd/apps/reader_node.go

@@ -67,6 +67,7 @@ func RegisterReaderNodeApp[B firecore.Block](chain *firecore.Chain[B], rootLog *
 			flags.Bool("reader-node-discard-after-stop-num", false, "Ignore remaining blocks being processed after stop num (only useful if we discard the reader data after reprocessing a chunk of blocks)")
 			flags.String("reader-node-working-dir", "{data-dir}/reader/work", "Path where reader will stores its files")
 			flags.Uint("reader-node-start-block-num", 0, "Blocks that were produced with smaller block number then the given block num are skipped")
+			flags.String("reader-node-start-block-timestamp", "", "Blocks that were produced before this timestamp are skipped (inclusive gate, supports unix seconds or RFC3339)")
 			flags.Uint("reader-node-stop-block-num", 0, "Shutdown reader when we the following 'stop-block-num' has been reached, inclusively.")
 			flags.Int("reader-node-blocks-chan-capacity", 100, "Capacity of the channel holding blocks read by the reader. Process will shutdown reader-node if the channel gets over 90% of that capacity to prevent horrible consequences. Raise this number when processing tiny blocks very quickly")
 			flags.Uint64("reader-node-line-buffer-size", 209715200, "Capacity of the buffer for reading a single line out of the node, in bytes (This is a hard limit. Some future enormouse blocks may require raising this to process them).")
@@ -148,6 +149,10 @@ func RegisterReaderNodeApp[B firecore.Block](chain *firecore.Chain[B], rootLog *
 			}
 
 			stopBlockNum := viper.GetUint64("reader-node-stop-block-num")
+			startBlockTimestamp, err := parseReaderNodeStartBlockTimestamp(viper.GetString("reader-node-start-block-timestamp"))
+			if err != nil {
+				return nil, err
+			}
 
 			hostname, _ := os.Hostname()
 			nodeArgumentResolver := createNodeArgumentsResolver(sfDataDir, nodeDataDir, hostname, firstStreamableBlock, resolveStartBlockNum, stopBlockNum)
@@ -230,6 +235,7 @@ func RegisterReaderNodeApp[B firecore.Block](chain *firecore.Chain[B], rootLog *
 					return chain.ConsoleReaderFactory(lines, chain.BlockEncoder, appLogger, appTracer)
 				},
 				resolveStartBlockNum,
+				startBlockTimestamp,
 				stopBlockNum,
 				blocksChanCapacity,
 				metricsAndReadinessManager.UpdateHeadBlock,

cmd/apps/reader_node_firehose.go

@@ -48,6 +48,10 @@ func RegisterReaderNodeFirehoseApp[B firecore.Block](chain *firecore.Chain[B], r
 		},
 		FactoryFunc: func(runtime *launcher.Runtime) (launcher.App, error) {
 			sfDataDir := runtime.AbsDataDir
+			startBlockTimestamp, err := parseReaderNodeStartBlockTimestamp(viper.GetString("reader-node-start-block-timestamp"))
+			if err != nil {
+				return nil, err
+			}
 
 			// Initialize test mode comparator if enabled
 			testModeComparator, err := createTestModeComparator(chain, appLogger)
@@ -70,6 +74,7 @@ func RegisterReaderNodeFirehoseApp[B firecore.Block](chain *firecore.Chain[B], r
 				OneBlocksStoreURL:   oneBlockStoreURL,
 				OneBlockSuffix:      viper.GetString("reader-node-one-block-suffix"),
 				StartBlockNum:       viper.GetUint64("reader-node-start-block-num"),
+				StartBlockTimestamp: startBlockTimestamp,
 				StopBlockNum:        viper.GetUint64("reader-node-stop-block-num"),
 				StateFile:           stateFile,
 				ReadinessMaxLatency: viper.GetDuration("reader-node-readiness-max-latency"),

cmd/apps/reader_node_start_block_timestamp.go

@@ -0,0 +1,29 @@
+package apps
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+)
+
+func parseReaderNodeStartBlockTimestamp(input string) (*time.Time, error) {
+	input = strings.TrimSpace(input)
+	if input == "" {
+		return nil, nil
+	}
+
+	if unixSeconds, err := strconv.ParseInt(input, 10, 64); err == nil {
+		t := time.Unix(unixSeconds, 0).UTC()
+		return &t, nil
+	}
+
+	for _, layout := range []string{time.RFC3339Nano, time.RFC3339} {
+		if parsed, err := time.Parse(layout, input); err == nil {
+			t := parsed.UTC()
+			return &t, nil
+		}
+	}
+
+	return nil, fmt.Errorf("invalid reader-node-start-block-timestamp %q: expected unix seconds or RFC3339", input)
+}

cmd/apps/reader_node_start_block_timestamp_test.go

@@ -0,0 +1,61 @@
+package apps
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_parseReaderNodeStartBlockTimestamp(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   string
+		want    *time.Time
+		wantErr string
+	}{
+		{
+			name:  "empty means no gate",
+			input: "",
+			want:  nil,
+		},
+		{
+			name:  "unix seconds",
+			input: "1712880000",
+			want:  ptrTime(time.Unix(1712880000, 0).UTC()),
+		},
+		{
+			name:  "rfc3339",
+			input: "2024-04-12T00:00:00Z",
+			want:  ptrTime(time.Date(2024, 4, 12, 0, 0, 0, 0, time.UTC)),
+		},
+		{
+			name:    "invalid",
+			input:   "not-a-timestamp",
+			wantErr: "invalid reader-node-start-block-timestamp",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := parseReaderNodeStartBlockTimestamp(test.input)
+			if test.wantErr != "" {
+				require.ErrorContains(t, err, test.wantErr)
+				return
+			}
+
+			require.NoError(t, err)
+			if test.want == nil {
+				require.Nil(t, got)
+				return
+			}
+
+			require.NotNil(t, got)
+			require.Equal(t, *test.want, *got)
+		})
+	}
+}
+
+func ptrTime(t time.Time) *time.Time {
+	return &t
+}

cmd/apps/reader_node_stdin.go

@@ -42,6 +42,10 @@ func RegisterReaderNodeStdinApp[B firecore.Block](chain *firecore.Chain[B], root
 		},
 		FactoryFunc: func(runtime *launcher.Runtime) (launcher.App, error) {
 			sfDataDir := runtime.AbsDataDir
+			startBlockTimestamp, err := parseReaderNodeStartBlockTimestamp(viper.GetString("reader-node-start-block-timestamp"))
+			if err != nil {
+				return nil, err
+			}
 
 			// Initialize test mode comparator if enabled
 			testModeComparator, err := createTestModeComparator(chain, appLogger)
@@ -70,6 +74,7 @@ func RegisterReaderNodeStdinApp[B firecore.Block](chain *firecore.Chain[B], root
 				OneBlocksStoreURL:          oneBlocksStoreURL,
 				MindReadBlocksChanCapacity: viper.GetInt("reader-node-blocks-chan-capacity"),
 				StartBlockNum:              viper.GetUint64("reader-node-start-block-num"),
+				StartBlockTimestamp:        startBlockTimestamp,
 				StopBlockNum:               viper.GetUint64("reader-node-stop-block-num"),
 				WorkingDir:                 firecore.MustReplaceDataDir(sfDataDir, viper.GetString("reader-node-working-dir")),
 				OneBlockSuffix:             viper.GetString("reader-node-one-block-suffix"),

node-manager/app/firehose_reader/app.go

@@ -39,6 +39,7 @@ type Config struct {
 	OneBlocksStoreURL   string
 	OneBlockSuffix      string
 	StartBlockNum       uint64
+	StartBlockTimestamp *time.Time
 	StopBlockNum        uint64
 	ReadinessMaxLatency time.Duration
 	StateFile           string

node-manager/app/firehose_reader/syncer.go

@@ -135,6 +135,14 @@ func (s *syncer) Run() error {
 				Payload:   response.Block,
 			}
 
+			if s.config.StartBlockTimestamp != nil && pbBlock.Time().Before(*s.config.StartBlockTimestamp) {
+				lastCursor = response.Cursor
+				if err := s.writeCursor(lastCursor); err != nil {
+					return fmt.Errorf("writing cursor: %w", err)
+				}
+				continue
+			}
+
 			// In test mode, compare blocks instead of writing them
 			if s.testModeComparator != nil {
 				if err := s.testModeComparator.CompareBlock(s.appCtx, pbBlock); err != nil {

node-manager/app/node_reader_stdin/app.go

@@ -18,6 +18,7 @@ import (
 	"bufio"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/streamingfast/bstream/blockstream"
 	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
@@ -41,6 +42,7 @@ type Config struct {
 	OneBlockSuffix             string
 	MindReadBlocksChanCapacity int
 	StartBlockNum              uint64
+	StartBlockTimestamp        *time.Time
 	StopBlockNum               uint64
 	WorkingDir                 string
 	LogToZap                   bool
@@ -121,6 +123,7 @@ func (a *App) Run() error {
 		a.Config.WorkingDir,
 		a.modules.ConsoleReaderFactory,
 		a.Config.StartBlockNum,
+		a.Config.StartBlockTimestamp,
 		a.Config.StopBlockNum,
 		a.Config.MindReadBlocksChanCapacity,
 		a.modules.MetricsAndReadinessManager.UpdateHeadBlock,

node-manager/mindreader/mindreader.go

@@ -63,6 +63,7 @@ type MindReaderPlugin struct {
 
 	archiver                 *Archiver // transformed blocks are sent to Archiver
 	consoleReaderFactory     ConsolerReaderFactory
+	startBlockTimestamp      *time.Time
 	stopBlock                uint64 // if set, call shutdownFunc(nil) when we hit this number
 	channelCapacity          int    // transformed blocks are buffered in a channel
 	forceFinalityAfterBlocks *uint64
@@ -98,6 +99,7 @@ func NewMindReaderPlugin(
 	workingDirectory string,
 	consoleReaderFactory ConsolerReaderFactory,
 	startBlockNum uint64,
+	startBlockTimestamp *time.Time,
 	stopBlockNum uint64,
 	channelCapacity int,
 	headBlockUpdater nodeManager.HeadBlockUpdater,
@@ -120,6 +122,7 @@ func NewMindReaderPlugin(
 		zap.String("one_block_suffix", oneBlockSuffix),
 		zap.String("working_directory", workingDirectory),
 		zap.Uint64("start_block_num", startBlockNum),
+		zap.Timep("start_block_timestamp", startBlockTimestamp),
 		zap.Uint64("stop_block_num", stopBlockNum),
 		zap.Int("channel_capacity", channelCapacity),
 		zap.Bool("with_head_block_updater", headBlockUpdater != nil),
@@ -164,6 +167,7 @@ func NewMindReaderPlugin(
 		Shutter:                  shutter.New(),
 		archiver:                 archiver,
 		consoleReaderFactory:     consoleReaderFactory,
+		startBlockTimestamp:      startBlockTimestamp,
 		stopBlock:                stopBlockNum,
 		channelCapacity:          channelCapacity,
 		headBlockUpdater:         headBlockUpdater,
@@ -460,6 +464,10 @@ func (p *MindReaderPlugin) readOneMessage(blocks chan<- *pbbstream.Block) error
 		return nil
 	}
 
+	if p.startBlockTimestamp != nil && block.Time().Before(*p.startBlockTimestamp) {
+		return nil
+	}
+
 	p.lastSeenBlockLock.Lock()
 	p.lastSeenBlock = block.AsRef()
 	p.lastSeenBlockLock.Unlock()

node-manager/mindreader/mindreader_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/streamingfast/shutter"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func TestMindReaderPlugin_OfficialPrefix_ReadFlow(t *testing.T) {
@@ -108,6 +109,38 @@ func TestMindReaderPlugin_StopAtBlockNumReached(t *testing.T) {
 	assert.Equal(t, numOfLines, len(blocks)) // moderate requirement, race condition can make it pass more blocks
 }
 
+func TestMindReaderPlugin_StartBlockTimestampReached(t *testing.T) {
+	lines := make(chan string, 2)
+	blocks := make(chan *pbbstream.Block, 2)
+
+	startBlockTimestamp := time.Unix(1000, 0).UTC()
+	mindReader := &MindReaderPlugin{
+		Shutter:             shutter.New(),
+		lines:               lines,
+		consoleReader:       newTestConsoleReader(lines),
+		startBlockTimestamp: &startBlockTimestamp,
+	}
+
+	blockBeforeStartTime := startBlockTimestamp.Add(-time.Second)
+	blockAtStartTime := startBlockTimestamp
+
+	mindReader.LogLine(fmt.Sprintf(`DMLOG {"id":"00000001a","timestamp":"%s"}`, blockBeforeStartTime.Format(time.RFC3339Nano)))
+	err := mindReader.readOneMessage(blocks)
+	require.NoError(t, err)
+	assert.Equal(t, 0, len(blocks))
+
+	mindReader.LogLine(fmt.Sprintf(`DMLOG {"id":"00000002a","timestamp":"%s"}`, blockAtStartTime.Format(time.RFC3339Nano)))
+	err = mindReader.readOneMessage(blocks)
+	require.NoError(t, err)
+
+	select {
+	case block := <-blocks:
+		require.Equal(t, uint64(2), block.Number)
+	case <-time.After(time.Second):
+		t.Fatal("expected block to pass the timestamp gate")
+	}
+}
+
 func TestMindReaderPlugin_OneBlockSuffixFormat(t *testing.T) {
 	assert.Error(t, validateOneBlockSuffix(""))
 	assert.NoError(t, validateOneBlockSuffix("example"))
@@ -142,17 +175,28 @@ func (c *testConsoleReader) ReadBlock() (*pbbstream.Block, error) {
 	}
 
 	type block struct {
-		ID string `json:"id"`
+		ID        string `json:"id"`
+		Timestamp string `json:"timestamp,omitempty"`
 	}
 
 	data := new(block)
 	if err := json.Unmarshal([]byte(formatedLine), data); err != nil {
 		return nil, fmt.Errorf("marshalling error on '%s': %w", formatedLine, err)
 	}
-	return &pbbstream.Block{
+	out := &pbbstream.Block{
 		Id:     data.ID,
 		Number: toBlockNum(data.ID),
-	}, nil
+	}
+
+	if data.Timestamp != "" {
+		timestamp, err := time.Parse(time.RFC3339Nano, data.Timestamp)
+		if err != nil {
+			return nil, fmt.Errorf("invalid timestamp %q: %w", data.Timestamp, err)
+		}
+		out.Timestamp = timestamppb.New(timestamp)
+	}
+
+	return out, nil
 }
 
 func toBlockNum(blockID string) uint64 {