fix(logpoller): replay lower boundary revalidation (#603)

* fix: replay lower boundary revalidation

* chore: clean up

* fix: check BlockIDExt

* chore: lint

Author: jadepark-dev

pkg/logpoller/o11y_log_provider.go

@@ -2,6 +2,7 @@ package logpoller
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 
@@ -30,6 +31,10 @@ func NewTonO11yLogProvider(client ton.APIClientWrapped, loader TxLoader) RawLogP
 
 // GetLogs retrieves all ExternalMsgOutLogs for an address between fromBlockSeqNo (exclusive) and toBlock (inclusive).
 func (tlp *tonO11yLogProvider) GetLogs(ctx context.Context, addr *address.Address, from uint32, to *ton.BlockIDExt) ([]models.RawLog, error) {
+	if to == nil {
+		return nil, errors.New("to block must not be nil")
+	}
+
 	// validate that the provided block belongs to the masterchain
 	if to.Workchain != address.MasterchainID {
 		return nil, fmt.Errorf("expected masterchain block (workchain %d), got workchain %d", address.MasterchainID, to.Workchain)

pkg/logpoller/replay.go

@@ -133,8 +133,14 @@ func (lp *service) applyReplayOverride(ctx context.Context, blockRange *models.B
 		return blockRange, 0 // No replay request
 	}
 
-	// re-validate that the replay target block still exists
-	_, err := lp.lookupRequestedReplayBlock(ctx, fromBlock, currentMasterchainBlock)
+	// re-validate prevBlock (fromBlock-1) to handle the edge case where it gets pruned between
+	// request and override time. Validate fromBlock-1 since it's the block used by
+	// GetTransactionLTBounds for startLT.
+	blockToValidate := fromBlock // fromBlock == 1: no block 0 on TON (localnet edge case)
+	if fromBlock > 1 {
+		blockToValidate = fromBlock - 1
+	}
+	_, err := lp.lookupRequestedReplayBlock(ctx, blockToValidate, currentMasterchainBlock)
 	if err != nil {
 		lp.lggr.Warnw("replay rejected", "error", err, "fromBlock", fromBlock)
 		lp.clearReplayRequest()

pkg/logpoller/replay_test.go

@@ -48,8 +48,8 @@ func TestApplyReplayOverride(t *testing.T) {
 		lp := &service{
 			lggr: logger.Sugared(logger.Nop()),
 			clientProvider: func(_ context.Context) (ton.APIClientWrapped, error) {
-				// Validation looks up replayFromBlock (51)
-				return &mockAPIClient{lookupBlockResult: &ton.BlockIDExt{SeqNo: 51}}, nil
+				// Validation looks up prevBlock (50)
+				return &mockAPIClient{lookupBlockResult: &ton.BlockIDExt{SeqNo: 50}}, nil
 			},
 		}
 		lp.replay.status = models.ReplayStatusRequested
@@ -77,8 +77,8 @@ func TestApplyReplayOverride(t *testing.T) {
 		lp := &service{
 			lggr: logger.Sugared(logger.Nop()),
 			clientProvider: func(_ context.Context) (ton.APIClientWrapped, error) {
-				// Validation looks up replayFromBlock (51)
-				return &mockAPIClient{lookupBlockResult: &ton.BlockIDExt{SeqNo: 51}}, nil
+				// Validation looks up prevBlock (50)
+				return &mockAPIClient{lookupBlockResult: &ton.BlockIDExt{SeqNo: 50}}, nil
 			},
 		}
 		lp.replay.status = models.ReplayStatusRequested
@@ -94,7 +94,7 @@ func TestApplyReplayOverride(t *testing.T) {
 	t.Run("replay rejected and status reset when block beyond current", func(t *testing.T) {
 		t.Parallel()
 		currentMasterchainBlock := &ton.BlockIDExt{Workchain: address.MasterchainID, SeqNo: 100, Shard: 1}
-		// Replay from block 151 stores fromBlock=151, prevBlock=150, validation checks 151 >= 100
+		// Replay from block 151 stores fromBlock=151, prevBlock=150, validation checks 150 >= 100
 		prevBlock := &ton.BlockIDExt{Workchain: address.MasterchainID, SeqNo: 150, Shard: 1}
 
 		lp := &service{
@@ -115,25 +115,26 @@ func TestApplyReplayOverride(t *testing.T) {
 		require.Nil(t, lp.replay.prevBlock)
 	})
 
-	t.Run("replay rejected and status reset when block pruned", func(t *testing.T) {
+	t.Run("replay rejected when prevBlock pruned between request and override", func(t *testing.T) {
 		t.Parallel()
 		currentMasterchainBlock := &ton.BlockIDExt{Workchain: address.MasterchainID, SeqNo: 100, Shard: 1}
-		// Replay from block 51 stores fromBlock=51, prevBlock=50, validation looks up block 51
+		// Replay from block 51 stores fromBlock=51, prevBlock=50, validation looks up prevBlock (50)
 		prevBlock := &ton.BlockIDExt{Workchain: address.MasterchainID, SeqNo: 50, Shard: 1}
 
 		lp := &service{
 			lggr: logger.Sugared(logger.Nop()),
 			clientProvider: func(_ context.Context) (ton.APIClientWrapped, error) {
+				// prevBlock (50) was pruned since the original Replay() request
 				return &mockAPIClient{lookupBlockErr: ton.ErrBlockNotFound}, nil
 			},
 		}
 		lp.replay.status = models.ReplayStatusRequested
 		lp.replay.fromBlock = 51
 		lp.replay.prevBlock = prevBlock
 
-		result, _ := lp.applyReplayOverride(context.Background(), nil, currentMasterchainBlock)
+		result, replayFrom := lp.applyReplayOverride(context.Background(), nil, currentMasterchainBlock)
 		require.Nil(t, result)
-		// Status should be NoRequest after clearReplayRequest() is called (rejection resets to initial state)
+		require.Equal(t, uint32(0), replayFrom)
 		require.Equal(t, models.ReplayStatusNoRequest, lp.replay.status)
 		require.Equal(t, uint32(0), lp.replay.fromBlock)
 		require.Nil(t, lp.replay.prevBlock)
@@ -146,7 +147,7 @@ func TestApplyReplayOverride(t *testing.T) {
 		lp := &service{
 			lggr: logger.Sugared(logger.Nop()),
 			clientProvider: func(_ context.Context) (ton.APIClientWrapped, error) {
-				// Validation looks up fromBlock (1)
+				// fromBlock == 1: no block 0 on TON, so validation falls back to block 1
 				return &mockAPIClient{lookupBlockResult: &ton.BlockIDExt{SeqNo: 1}}, nil
 			},
 		}