chore: ignore captures taken before a market's settle time

extensions/tn_settlement/README.md

@@ -54,7 +54,7 @@ Every N minutes (configurable):
   │         LIMIT max_markets_per_run
   │
   ├─ For each market:
-  │  ├─ Check if attestation exists AND is signed
+  │  ├─ Check for a signed capture taken at or after settle_time
   │  ├─ If yes → Broadcast settle_market(query_id) transaction
   │  ├─ Retry up to N times with exponential backoff
   │  └─ Log success/failure
@@ -72,7 +72,7 @@ settle_market($query_id)
 
 The action:
 1. Validates settlement time has arrived
-2. Retrieves signed attestation for market hash
+2. Retrieves the earliest signed attestation for the market hash taken at or after settle_time
 3. Parses result to determine winning outcome
 4. Updates market: `settled=TRUE, winning_outcome=result, settled_at=timestamp`
 
@@ -116,7 +116,7 @@ INFO: tn_settlement (re)started with new schedule, schedule=*/10 * * * *
 
 ### Engine Operations (`internal/engine_ops.go`)
 - `FindUnsettledMarkets()` - Query markets ready for settlement
-- `AttestationExists()` - Check if signed attestation available
+- `AttestationExists()` - Check if a signed attestation taken at or after settle_time is available
 - `BroadcastSettleMarketWithRetry()` - Build, sign, broadcast transaction
 
 ### Lifecycle Management (`scheduler_lifecycle.go`)

extensions/tn_settlement/internal/engine_ops.go

@@ -10,6 +10,7 @@ import (
 
 	gethAbi "github.com/ethereum/go-ethereum/accounts/abi"
 	gethCommon "github.com/ethereum/go-ethereum/common"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/trufnetwork/kwil-db/common"
 	"github.com/trufnetwork/kwil-db/core/crypto/auth"
 	"github.com/trufnetwork/kwil-db/core/log"
@@ -34,6 +35,10 @@ type EngineOperations struct {
 	readDB   sql.DB                 // Independent read handle for poll reads; bypasses the engine interpreter lock
 	accounts common.Accounts
 
+	// untimedCapturesOnce logs, once, that attestations has no created_timestamp
+	// yet and every capture is being counted. See CaptureStatusFor.
+	untimedCapturesOnce sync.Once
+
 	// nonceMu guards the settlement cycle's nonce counter below.
 	nonceMu sync.Mutex
 	// cycleNonce is the nonce for this cycle's next transaction, and cycleNonceSet
@@ -305,10 +310,16 @@ type CaptureStatus struct {
 	CapturedAt int64
 }
 
-// CaptureStatusFor reports what has been captured for a market.
+// CaptureStatusFor reports what has been captured for a market at or after its
+// settle time.
+//
+// A capture taken before settleTime does not count, and neither does one with no
+// recorded time: settle_market never resolves a market on either. Counting one
+// would leave the market waiting on a capture that can never settle it, when what
+// it needs is a new capture.
 //
 // Reads run on the independent readDB handle rather than the engine interpreter.
-func (e *EngineOperations) CaptureStatusFor(ctx context.Context, marketHash []byte) (CaptureStatus, error) {
+func (e *EngineOperations) CaptureStatusFor(ctx context.Context, marketHash []byte, settleTime int64) (CaptureStatus, error) {
 	if e.readDB == nil {
 		return CaptureStatus{}, fmt.Errorf("settlement read handle not initialized")
 	}
@@ -319,8 +330,24 @@ func (e *EngineOperations) CaptureStatusFor(ctx context.Context, marketHash []by
 		`SELECT bool_or(signature IS NOT NULL) AS signed,
 		        min(created_height)           AS captured_at
 		   FROM main.attestations
-		  WHERE attestation_hash = $1`,
-		marketHash)
+		  WHERE attestation_hash = $1
+		    AND created_timestamp >= $2`,
+		marketHash, settleTime)
+	if isUndefinedColumn(err) {
+		// The binary is ahead of its release's SQL. Until migrate.sh adds
+		// created_timestamp, settle_market is still the version that takes any
+		// signed capture, so count every capture as both did before rather than
+		// fail the check for every market and settle nothing.
+		e.untimedCapturesOnce.Do(func() {
+			e.logger.Warn("attestations has no created_timestamp yet; counting captures taken at any time until the migrations are applied")
+		})
+		rs, err = e.readDB.Execute(ctx,
+			`SELECT bool_or(signature IS NOT NULL) AS signed,
+			        min(created_height)           AS captured_at
+			   FROM main.attestations
+			  WHERE attestation_hash = $1`,
+			marketHash)
+	}
 	if err != nil {
 		return CaptureStatus{}, fmt.Errorf("check attestation: %w", err)
 	}
@@ -339,17 +366,29 @@ func (e *EngineOperations) CaptureStatusFor(ctx context.Context, marketHash []by
 	}, nil
 }
 
-// AttestationExists checks if a signed attestation exists for the given hash.
+// AttestationExists checks if a signed attestation taken at or after settleTime
+// exists for the given hash.
 //
 // Reads run on the independent readDB handle rather than the engine interpreter.
-func (e *EngineOperations) AttestationExists(ctx context.Context, marketHash []byte) (bool, error) {
-	status, err := e.CaptureStatusFor(ctx, marketHash)
+func (e *EngineOperations) AttestationExists(ctx context.Context, marketHash []byte, settleTime int64) (bool, error) {
+	status, err := e.CaptureStatusFor(ctx, marketHash, settleTime)
 	if err != nil {
 		return false, err
 	}
 	return status.Signed, nil
 }
 
+// undefinedColumnCode is the Postgres SQLSTATE for a reference to a column that
+// does not exist.
+const undefinedColumnCode = "42703"
+
+// isUndefinedColumn reports whether err is Postgres refusing a statement that
+// names a column the table does not have.
+func isUndefinedColumn(err error) bool {
+	var pgErr *pgconn.PgError
+	return errors.As(err, &pgErr) && pgErr.Code == undefinedColumnCode
+}
+
 // BroadcastSettleMarketWithRetry broadcasts settle_market transaction with retry logic.
 // Uses exponential backoff since external systems may use the same wallet (nonce conflicts).
 func (e *EngineOperations) BroadcastSettleMarketWithRetry(

extensions/tn_settlement/internal/engine_ops_test.go

@@ -12,6 +12,7 @@ import (
 
 	gethAbi "github.com/ethereum/go-ethereum/accounts/abi"
 	gethCommon "github.com/ethereum/go-ethereum/common"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/stretchr/testify/require"
 	"github.com/trufnetwork/kwil-db/core/crypto"
 	"github.com/trufnetwork/kwil-db/core/crypto/auth"
@@ -36,7 +37,7 @@ func TestPollReads_FailClosedWhenReadDBNil(t *testing.T) {
 	_, err = ops.FindUnsettledMarkets(ctx, 10)
 	require.ErrorContains(t, err, want, "FindUnsettledMarkets must fail closed")
 
-	_, err = ops.AttestationExists(ctx, []byte{0x01})
+	_, err = ops.AttestationExists(ctx, []byte{0x01}, 1)
 	require.ErrorContains(t, err, want, "AttestationExists must fail closed")
 
 	_, err = ops.GetMarketQueryComponents(ctx, 1)
@@ -959,14 +960,98 @@ func TestCaptureStatusFor_SeparatesCapturedFromSigned(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			ops := &EngineOperations{logger: log.DiscardLogger, readDB: &stubReadDB{row: tc.row}}
 
-			got, err := ops.CaptureStatusFor(context.Background(), []byte{0xab})
+			got, err := ops.CaptureStatusFor(context.Background(), []byte{0xab}, 1)
 			require.NoError(t, err)
 			require.Equal(t, tc.want, got)
 
 			// AttestationExists keeps its old meaning: signed, not merely present.
-			exists, err := ops.AttestationExists(context.Background(), []byte{0xab})
+			exists, err := ops.AttestationExists(context.Background(), []byte{0xab}, 1)
 			require.NoError(t, err)
 			require.Equal(t, tc.want.Signed, exists)
 		})
 	}
 }
+
+// =============================================================================
+// Test: capture status only counts captures taken at or after settle time
+// =============================================================================
+
+// recordingReadDB records every statement and its arguments. A statement that
+// names created_timestamp fails with failTimed when that is set, as on a database
+// whose attestations table predates the column.
+type recordingReadDB struct {
+	row       []any
+	failTimed error
+	stmts     []string
+	args      [][]any
+}
+
+var _ sql.DB = (*recordingReadDB)(nil)
+
+func (r *recordingReadDB) Execute(ctx context.Context, stmt string, args ...any) (*sql.ResultSet, error) {
+	r.stmts = append(r.stmts, stmt)
+	r.args = append(r.args, args)
+	if r.failTimed != nil && strings.Contains(stmt, "created_timestamp") {
+		return nil, fmt.Errorf("execute query: %w", r.failTimed)
+	}
+	return &sql.ResultSet{
+		Columns: []string{"signed", "captured_at"},
+		Rows:    [][]any{r.row},
+	}, nil
+}
+
+func (r *recordingReadDB) BeginTx(ctx context.Context) (sql.Tx, error) {
+	return nil, fmt.Errorf("recording read handle does not support transactions")
+}
+
+// settle_market resolves a market only on a capture taken at or after its
+// settle_time (node#1435). The status the scheduler acts on has to apply the same
+// rule, or a market whose only capture is early looks captured, is never captured
+// again, and waits forever on a capture settle_market refuses.
+func TestCaptureStatusFor_CountsOnlyCapturesTakenAtOrAfterSettleTime(t *testing.T) {
+	db := &recordingReadDB{row: []any{true, int64(2615513)}}
+	ops := &EngineOperations{logger: log.DiscardLogger, readDB: db}
+
+	got, err := ops.CaptureStatusFor(context.Background(), []byte{0xab}, 1700000000)
+	require.NoError(t, err)
+	require.Equal(t, CaptureStatus{Captured: true, Signed: true, CapturedAt: 2615513}, got)
+
+	require.Len(t, db.stmts, 1)
+	require.Contains(t, db.stmts[0], "created_timestamp >= $2",
+		"the status must exclude captures taken before the settle time, and ones with no time")
+	require.Equal(t, []any{[]byte{0xab}, int64(1700000000)}, db.args[0])
+}
+
+// A release ships its binary before migrate.sh applies its SQL. Until then
+// attestations has no created_timestamp and settle_market still takes any signed
+// capture, so the status falls back to counting every capture rather than fail
+// for every market and settle nothing.
+func TestCaptureStatusFor_CountsEveryCaptureUntilTheColumnExists(t *testing.T) {
+	db := &recordingReadDB{
+		row:       []any{true, int64(2615513)},
+		failTimed: &pgconn.PgError{Code: "42703", Message: `column "created_timestamp" does not exist`},
+	}
+	ops := &EngineOperations{logger: log.DiscardLogger, readDB: db}
+
+	got, err := ops.CaptureStatusFor(context.Background(), []byte{0xab}, 1700000000)
+	require.NoError(t, err)
+	require.Equal(t, CaptureStatus{Captured: true, Signed: true, CapturedAt: 2615513}, got)
+
+	require.Len(t, db.stmts, 2, "one timed attempt, then one fallback")
+	require.NotContains(t, db.stmts[1], "created_timestamp")
+	require.Equal(t, []any{[]byte{0xab}}, db.args[1])
+}
+
+// Only a missing column falls back. Any other failure, a lock timeout above all,
+// keeps failing the check closed, as every poll read does.
+func TestCaptureStatusFor_OtherErrorsDoNotFallBack(t *testing.T) {
+	db := &recordingReadDB{
+		row:       []any{true, int64(2615513)},
+		failTimed: &pgconn.PgError{Code: "55P03", Message: "canceling statement due to lock timeout"},
+	}
+	ops := &EngineOperations{logger: log.DiscardLogger, readDB: db}
+
+	_, err := ops.CaptureStatusFor(context.Background(), []byte{0xab}, 1700000000)
+	require.ErrorContains(t, err, "lock timeout")
+	require.Len(t, db.stmts, 1, "no fallback read after a failure that is not a missing column")
+}

extensions/tn_settlement/scheduler/scheduler.go

@@ -36,7 +36,9 @@ type ProcessingGuard interface {
 // This interface allows for mocking in tests.
 type EngineOps interface {
 	FindUnsettledMarkets(ctx context.Context, limit int) ([]*internal.UnsettledMarket, error)
-	CaptureStatusFor(ctx context.Context, marketHash []byte) (internal.CaptureStatus, error)
+	// CaptureStatusFor counts only captures taken at or after settleTime, the
+	// only ones settle_market resolves a market on.
+	CaptureStatusFor(ctx context.Context, marketHash []byte, settleTime int64) (internal.CaptureStatus, error)
 	// BeginCycle discards the nonce counter left over from the previous cycle so
 	// the next transaction re-seeds from committed account state.
 	BeginCycle()
@@ -350,7 +352,7 @@ func (s *SettlementScheduler) runSettlementCycle(
 
 		processed++
 
-		capture, attErr := engineOps.CaptureStatusFor(ctx, market.Hash)
+		capture, attErr := engineOps.CaptureStatusFor(ctx, market.Hash, market.SettleTime)
 		if attErr != nil {
 			s.logger.Warn("failed to check attestation",
 				"query_id", market.ID,

extensions/tn_settlement/scheduler/scheduler_test.go

@@ -121,6 +121,9 @@ type mockEngineOps struct {
 
 	beginCycleCalls int
 	requestCalls    []int
+	// captureSettleTimes records the settle time each capture-status check was
+	// asked about, in order.
+	captureSettleTimes []int64
 	// calls records every broadcast the cycle made, in order, as "capture:<id>"
 	// or "settle:<id>". It is what proves no settlement landed between two
 	// captures.
@@ -147,7 +150,8 @@ func (m *mockEngineOps) FindUnsettledMarkets(ctx context.Context, limit int) ([]
 	return markets, nil
 }
 
-func (m *mockEngineOps) CaptureStatusFor(ctx context.Context, marketHash []byte) (internal.CaptureStatus, error) {
+func (m *mockEngineOps) CaptureStatusFor(ctx context.Context, marketHash []byte, settleTime int64) (internal.CaptureStatus, error) {
+	m.captureSettleTimes = append(m.captureSettleTimes, settleTime)
 	signed := m.attestationExists
 	captured := m.attestationExists
 	if m.attestedFor != nil && len(marketHash) > 0 {
@@ -819,3 +823,23 @@ func TestRunSettlementCycle_AwaitingSignatureIsNotCapturedAgain(t *testing.T) {
 		"market 1 is already captured and must be neither re-captured nor settled")
 	require.Equal(t, []int{2}, ops.requestCalls)
 }
+
+// A market's capture status is asked about at that market's own settle time. Only
+// a capture taken at or after it can settle the market, so a status checked
+// against any other time either counts a capture settle_market will refuse, and
+// the market waits forever, or misses one it would accept.
+func TestRunSettlementCycle_ChecksCapturesAtEachMarketsSettleTime(t *testing.T) {
+	ops := &mockEngineOps{
+		markets: []*internal.UnsettledMarket{
+			{ID: 1, Hash: []byte{1}, SettleTime: 1700000000},
+			{ID: 2, Hash: []byte{2}, SettleTime: 1700003600},
+			{ID: 3, Hash: []byte{3}, SettleTime: 1700007200},
+		},
+		attestedFor: map[int]bool{1: true},
+	}
+	s := mixedCycleScheduler(t, ops)
+
+	require.NoError(t, s.RunOnce(context.Background()))
+
+	require.Equal(t, []int64{1700000000, 1700003600, 1700007200}, ops.captureSettleTimes)
+}