From 772eeb801e7891fb86b086aeb287e9846ae37e73 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Apr 2026 20:15:45 +0000 Subject: [PATCH 1/3] Initial plan From f718a4b7d904cb12602d24e923768ec10a1b0029 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Apr 2026 20:24:20 +0000 Subject: [PATCH 2/3] Add reader-node start block timestamp gate Agent-Logs-Url: https://github.com/streamingfast/firehose-core/sessions/19007e17-dd35-4c28-a1f9-0e75b3dafa0e Co-authored-by: maoueh <123014+maoueh@users.noreply.github.com> --- cmd/apps/reader_node.go | 6 ++ cmd/apps/reader_node_firehose.go | 5 ++ cmd/apps/reader_node_start_block_timestamp.go | 29 +++++++++ .../reader_node_start_block_timestamp_test.go | 61 +++++++++++++++++++ cmd/apps/reader_node_stdin.go | 5 ++ node-manager/app/firehose_reader/app.go | 1 + node-manager/app/firehose_reader/syncer.go | 8 +++ node-manager/app/node_reader_stdin/app.go | 3 + node-manager/mindreader/mindreader.go | 8 +++ node-manager/mindreader/mindreader_test.go | 47 +++++++++++++- 10 files changed, 170 insertions(+), 3 deletions(-) create mode 100644 cmd/apps/reader_node_start_block_timestamp.go create mode 100644 cmd/apps/reader_node_start_block_timestamp_test.go diff --git a/cmd/apps/reader_node.go b/cmd/apps/reader_node.go index 8ac6c01..2e158c4 100644 --- a/cmd/apps/reader_node.go +++ b/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, diff --git a/cmd/apps/reader_node_firehose.go b/cmd/apps/reader_node_firehose.go index 158620e..40f6476 100644 --- a/cmd/apps/reader_node_firehose.go +++ b/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"), diff --git a/cmd/apps/reader_node_start_block_timestamp.go b/cmd/apps/reader_node_start_block_timestamp.go new file mode 100644 index 0000000..a5e0765 --- /dev/null +++ b/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) +} diff --git a/cmd/apps/reader_node_start_block_timestamp_test.go b/cmd/apps/reader_node_start_block_timestamp_test.go new file mode 100644 index 0000000..9a66086 --- /dev/null +++ b/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 +} diff --git a/cmd/apps/reader_node_stdin.go b/cmd/apps/reader_node_stdin.go index 277b64e..8676bab 100644 --- a/cmd/apps/reader_node_stdin.go +++ b/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"), diff --git a/node-manager/app/firehose_reader/app.go b/node-manager/app/firehose_reader/app.go index fed620e..6aa6740 100644 --- a/node-manager/app/firehose_reader/app.go +++ b/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 diff --git a/node-manager/app/firehose_reader/syncer.go b/node-manager/app/firehose_reader/syncer.go index d7e5690..90b9adc 100644 --- a/node-manager/app/firehose_reader/syncer.go +++ b/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 { diff --git a/node-manager/app/node_reader_stdin/app.go b/node-manager/app/node_reader_stdin/app.go index dcadd70..a0c2d28 100644 --- a/node-manager/app/node_reader_stdin/app.go +++ b/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, diff --git a/node-manager/mindreader/mindreader.go b/node-manager/mindreader/mindreader.go index fc674be..861497a 100644 --- a/node-manager/mindreader/mindreader.go +++ b/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() diff --git a/node-manager/mindreader/mindreader_test.go b/node-manager/mindreader/mindreader_test.go index add117d..5ada11d 100644 --- a/node-manager/mindreader/mindreader_test.go +++ b/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,35 @@ 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, + } + + mindReader.LogLine(`DMLOG {"id":"00000001a","timestamp":"1970-01-01T00:16:39Z"}`) + err := mindReader.readOneMessage(blocks) + require.NoError(t, err) + assert.Equal(t, 0, len(blocks)) + + mindReader.LogLine(`DMLOG {"id":"00000002a","timestamp":"1970-01-01T00:16:40Z"}`) + 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 +172,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 { From 2744e2ae60d741f2d714e8688ca9c5539ccd04ac Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Apr 2026 20:28:09 +0000 Subject: [PATCH 3/3] Refine timestamp gate test readability Agent-Logs-Url: https://github.com/streamingfast/firehose-core/sessions/19007e17-dd35-4c28-a1f9-0e75b3dafa0e Co-authored-by: maoueh <123014+maoueh@users.noreply.github.com> --- node-manager/mindreader/mindreader_test.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/node-manager/mindreader/mindreader_test.go b/node-manager/mindreader/mindreader_test.go index 5ada11d..d596684 100644 --- a/node-manager/mindreader/mindreader_test.go +++ b/node-manager/mindreader/mindreader_test.go @@ -121,12 +121,15 @@ func TestMindReaderPlugin_StartBlockTimestampReached(t *testing.T) { startBlockTimestamp: &startBlockTimestamp, } - mindReader.LogLine(`DMLOG {"id":"00000001a","timestamp":"1970-01-01T00:16:39Z"}`) + 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(`DMLOG {"id":"00000002a","timestamp":"1970-01-01T00:16:40Z"}`) + mindReader.LogLine(fmt.Sprintf(`DMLOG {"id":"00000002a","timestamp":"%s"}`, blockAtStartTime.Format(time.RFC3339Nano))) err = mindReader.readOneMessage(blocks) require.NoError(t, err)