From 62bc2dd36344362c3da568b03c308402fc5e7fc4 Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Wed, 7 Oct 2026 19:36:19 +0700 Subject: [PATCH 1/2] chore: ignore captures taken before a market's settle time --- extensions/tn_settlement/README.md | 6 +- .../tn_settlement/internal/engine_ops.go | 51 +++- .../tn_settlement/internal/engine_ops_test.go | 91 +++++- .../tn_settlement/scheduler/scheduler.go | 6 +- .../tn_settlement/scheduler/scheduler_test.go | 26 +- .../settlement_integration_test.go | 265 +++++++++++++++++- .../migrations/023-attestation-schema.sql | 12 + .../024-attestation-actions.prod.sql | 12 +- .../migrations/024-attestation-actions.sql | 12 +- .../migrations/032-order-book-actions.sql | 46 ++- tests/streams/order_book/order_events_test.go | 6 +- .../order_book/settlement_payout_test.go | 6 +- tests/streams/order_book/settlement_test.go | 19 +- 13 files changed, 511 insertions(+), 47 deletions(-) diff --git a/extensions/tn_settlement/README.md b/extensions/tn_settlement/README.md index f6ca0c40f..3ee0af641 100644 --- a/extensions/tn_settlement/README.md +++ b/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`) diff --git a/extensions/tn_settlement/internal/engine_ops.go b/extensions/tn_settlement/internal/engine_ops.go index 2def503fb..385c19564 100644 --- a/extensions/tn_settlement/internal/engine_ops.go +++ b/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,27 @@ 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" + +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( diff --git a/extensions/tn_settlement/internal/engine_ops_test.go b/extensions/tn_settlement/internal/engine_ops_test.go index 1e448a7f7..9da3d2c9f 100644 --- a/extensions/tn_settlement/internal/engine_ops_test.go +++ b/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") +} diff --git a/extensions/tn_settlement/scheduler/scheduler.go b/extensions/tn_settlement/scheduler/scheduler.go index ef18e523b..e25e06454 100644 --- a/extensions/tn_settlement/scheduler/scheduler.go +++ b/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, diff --git a/extensions/tn_settlement/scheduler/scheduler_test.go b/extensions/tn_settlement/scheduler/scheduler_test.go index 458254d0b..54864dc82 100644 --- a/extensions/tn_settlement/scheduler/scheduler_test.go +++ b/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:" // or "settle:". 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) +} diff --git a/extensions/tn_settlement/settlement_integration_test.go b/extensions/tn_settlement/settlement_integration_test.go index 09eaff36b..eb1def081 100644 --- a/extensions/tn_settlement/settlement_integration_test.go +++ b/extensions/tn_settlement/settlement_integration_test.go @@ -63,6 +63,8 @@ func TestSettlementIntegration(t *testing.T) { testSkipMarketWithoutAttestation(t), testMultipleMarketsProcessing(t), testSettleUsesTheEarliestCapture(t), + testSettleIgnoresACaptureTakenBeforeSettleTime(t), + testSettleIgnoresACaptureWithNoTime(t), }, }, testutils.GetTestOptionsWithCache()) } @@ -147,6 +149,11 @@ func testFindUnsettledMarkets(t *testing.T) func(context.Context, *kwilTesting.P } } +// captureTime is the block time the captures in this file are taken at. Every +// market here settles at 100, and settle_market resolves a market only on a +// capture taken at or after its settle time. +const captureTime = int64(150) + // ============================================================================= // Test: AttestationExists // ============================================================================= @@ -210,14 +217,14 @@ func testAttestationExists(t *testing.T) func(context.Context, *kwilTesting.Plat require.NoError(t, err) ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) - exists, err := ops.AttestationExists(ctx, attestationHash) + exists, err := ops.AttestationExists(ctx, attestationHash, 100) require.NoError(t, err) require.True(t, exists, "attestation should exist and be signed") // Test with non-existent hash fakeHash := make([]byte, 32) copy(fakeHash, []byte("nonexistent")) - exists, err = ops.AttestationExists(ctx, fakeHash) + exists, err = ops.AttestationExists(ctx, fakeHash, 100) require.NoError(t, err) require.False(t, exists, "fake attestation should not exist") @@ -225,13 +232,13 @@ func testAttestationExists(t *testing.T) func(context.Context, *kwilTesting.Plat // one, and it is the second — "has this been captured at all" — that // keeps a market from being captured twice. Exercised here rather than // against a stub so the aggregate SQL itself is covered. - status, err := ops.CaptureStatusFor(ctx, attestationHash) + status, err := ops.CaptureStatusFor(ctx, attestationHash, 100) require.NoError(t, err) require.True(t, status.Captured, "a signed capture is also a capture") require.True(t, status.Signed) require.NotZero(t, status.CapturedAt, "the capture height is what an operator watches") - status, err = ops.CaptureStatusFor(ctx, fakeHash) + status, err = ops.CaptureStatusFor(ctx, fakeHash, 100) require.NoError(t, err) require.Equal(t, internal.CaptureStatus{}, status, "no rows must decode to the zero status") @@ -245,6 +252,7 @@ func testAttestationExists(t *testing.T) func(context.Context, *kwilTesting.Plat var unsignedHash []byte engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime reqRes, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{deployer.Address(), streamID, "get_last_record", unsignedArgs, false, nil}, func(row *common.Row) error { @@ -254,13 +262,13 @@ func testAttestationExists(t *testing.T) func(context.Context, *kwilTesting.Plat require.NoError(t, err) require.NoError(t, reqRes.Error, "request_attestation failed") - status, err = ops.CaptureStatusFor(ctx, unsignedHash) + status, err = ops.CaptureStatusFor(ctx, unsignedHash, 100) require.NoError(t, err) require.True(t, status.Captured, "the row is on chain, so the market is captured") require.False(t, status.Signed, "nothing has signed it yet") require.NotZero(t, status.CapturedAt) - exists, err = ops.AttestationExists(ctx, unsignedHash) + exists, err = ops.AttestationExists(ctx, unsignedHash, 100) require.NoError(t, err) require.False(t, exists, "AttestationExists keeps its old meaning: signed, not merely present") @@ -453,7 +461,7 @@ func testSkipMarketWithoutAttestation(t *testing.T) func(context.Context, *kwilT require.NoError(t, err) ops := internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) - exists, err := ops.AttestationExists(ctx, attestationHash) + exists, err := ops.AttestationExists(ctx, attestationHash, settleTime) require.NoError(t, err) require.False(t, exists, "attestation without signature should not be considered ready") @@ -672,7 +680,8 @@ func createStreamWithoutSigningAttestation( }) require.NoError(t, err) - // Request attestation (but don't sign) + // Request attestation (but don't sign), after the market's settle time + engineCtx.TxContext.BlockContext.Timestamp = captureTime res, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -733,7 +742,9 @@ func createStreamAndAttestation( }) require.NoError(t, err) - // Request attestation using the SAME engineCtx so it sees the inserted data + // Request attestation using the SAME engineCtx so it sees the inserted data, + // at a block time after the settle time of every market in this file. + engineCtx.TxContext.BlockContext.Timestamp = captureTime var requestTxID string var attestationHash []byte res, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", @@ -833,6 +844,7 @@ func testSettleUsesTheEarliestCapture(t *testing.T) func(context.Context, *kwilT capture := func(height int64) []byte { ec := engineCtx ec.TxContext.BlockContext.Height = height + ec.TxContext.BlockContext.Timestamp = captureTime ec.TxContext.TxID = platform.Txid() var requestTxID string var hash []byte @@ -913,3 +925,238 @@ func testSettleUsesTheEarliestCapture(t *testing.T) func(context.Context, *kwilT return nil } } + +// ============================================================================= +// Test: a capture taken before settle_time never settles the market +// ============================================================================= + +// captureScenario is one market on one stream, with what the two tests below +// need to move the stream's value, capture its query at a chosen height and block +// time, and settle it. +type captureScenario struct { + t *testing.T + ctx context.Context + platform *kwilTesting.Platform + helper *attestationTests.AttestationTestHelper + engineCtx *common.EngineContext + ops *internal.EngineOperations + + dataProvider string + streamID string + argsBytes []byte + queryID int + marketHash []byte +} + +// newCaptureScenario creates a stream that accepts zeros and a market on its +// get_last_record query that settles at settleTime. Nothing is captured yet. +func newCaptureScenario(t *testing.T, ctx context.Context, platform *kwilTesting.Platform, + deployerHex, streamID string, settleTime int64) *captureScenario { + t.Helper() + lastTrufBalancePoint = nil + + deployer := util.Unsafe_NewEthereumAddressFromString(deployerHex) + platform.Deployer = deployer.Bytes() + + helper := attestationTests.NewAttestationTestHelper(t, ctx, platform) + require.NoError(t, setup.CreateDataProvider(ctx, platform, deployer.Address())) + require.NoError(t, erc20bridge.ForTestingInitializeExtension(ctx, platform)) + + // create_stream (100) + insert_records (1 each) + two captures (40 each). + require.NoError(t, giveTrufBalance(ctx, platform, deployer.Address(), "300000000000000000000")) + + sc := &captureScenario{ + t: t, ctx: ctx, platform: platform, helper: helper, + engineCtx: helper.NewEngineContext(), + dataProvider: deployer.Address(), + streamID: streamID, + } + + _, err := platform.Engine.Call(sc.engineCtx, platform.DB, "", "create_stream", + []any{streamID, "primitive"}, nil) + require.NoError(t, err) + + // The captures must disagree, and the second one reads a zero, which + // insert_records drops unless the stream opts in. + zerosRes, err := platform.Engine.Call(sc.engineCtx, platform.DB, "", "set_allow_zeros", + []any{sc.dataProvider, streamID, true}, nil) + require.NoError(t, err) + require.NoError(t, zerosRes.Error, "set_allow_zeros failed") + + sc.argsBytes, err = tn_utils.EncodeActionArgs([]any{ + sc.dataProvider, streamID, int64(1500), nil, false, + }) + require.NoError(t, err) + + queryComponents, err := encodeQueryComponents(sc.dataProvider, streamID, "get_last_record", sc.argsBytes) + require.NoError(t, err) + + createCtx := helper.NewEngineContext() + createCtx.TxContext.BlockContext.Timestamp = settleTime - 50 + createRes, err := platform.Engine.Call(createCtx, platform.DB, "", "create_market", + []any{testExtensionName, queryComponents, settleTime, int64(5), int64(1)}, + func(row *common.Row) error { + sc.queryID = int(row.Values[0].(int64)) + return nil + }) + require.NoError(t, err) + require.Nil(t, createRes.Error) + + require.NoError(t, platform.Engine.Execute(helper.NewEngineContext(), platform.DB, + `SELECT hash FROM ob_queries WHERE id = $id`, + map[string]any{"id": sc.queryID}, + func(row *common.Row) error { + sc.marketHash = append([]byte(nil), row.Values[0].([]byte)...) + return nil + })) + + accts, err := accounts.InitializeAccountStore(ctx, platform.DB, log.New()) + require.NoError(t, err) + sc.ops = internal.NewEngineOperations(platform.Engine, platform.DB, nil, platform.DB, accts, log.New()) + + return sc +} + +// insert records value at eventTime. parse_attestation_boolean reads a numeric +// action as value > 0, so 2 settles TRUE and 0 settles FALSE. +func (sc *captureScenario) insert(eventTime int64, value string) { + dec, err := kwilTypes.ParseDecimalExplicit(value, 36, 18) + require.NoError(sc.t, err) + res, err := sc.platform.Engine.Call(sc.engineCtx, sc.platform.DB, "", "insert_records", + []any{[]string{sc.dataProvider}, []string{sc.streamID}, []int64{eventTime}, []*kwilTypes.Decimal{dec}}, nil) + require.NoError(sc.t, err) + require.NoError(sc.t, res.Error) +} + +// capture takes and signs a capture of the market's query at height and block +// time blockTime. It runs on the scenario's one engine context, which is what +// lets it see the records inserted above. +func (sc *captureScenario) capture(height, blockTime int64) { + ec := sc.engineCtx + ec.TxContext.BlockContext.Height = height + ec.TxContext.BlockContext.Timestamp = blockTime + ec.TxContext.TxID = sc.platform.Txid() + var requestTxID string + var hash []byte + res, err := sc.platform.Engine.Call(ec, sc.platform.DB, "", "request_attestation", + []any{sc.dataProvider, sc.streamID, "get_last_record", sc.argsBytes, false, nil}, + func(row *common.Row) error { + requestTxID = row.Values[0].(string) + hash = append([]byte(nil), row.Values[1].([]byte)...) + return nil + }) + require.NoError(sc.t, err) + require.NoError(sc.t, res.Error, "request_attestation failed") + require.Equal(sc.t, sc.marketHash, hash, "every capture here is of the market's query") + sc.helper.SignAttestation(requestTxID) +} + +// settle calls settle_market at block time 200, after the market's settle time, +// and returns the action's error. +func (sc *captureScenario) settle() error { + ec := sc.helper.NewEngineContext() + ec.TxContext.BlockContext.Timestamp = 200 + res, err := sc.platform.Engine.Call(ec, sc.platform.DB, "", "settle_market", []any{sc.queryID}, nil) + require.NoError(sc.t, err) + return res.Error +} + +func (sc *captureScenario) winningOutcome() *bool { + var winning *bool + require.NoError(sc.t, sc.platform.Engine.Execute(sc.helper.NewEngineContext(), sc.platform.DB, + `SELECT winning_outcome FROM ob_queries WHERE id = $id`, + map[string]any{"id": sc.queryID}, + func(row *common.Row) error { + if row.Values[0] != nil { + v := row.Values[0].(bool) + winning = &v + } + return nil + })) + return winning +} + +// request_attestation is PUBLIC and a live market's query_components are +// readable, so anyone can capture a market's query before its settle_time, while +// the window it resolves on is still open. Ordered earliest first, that capture +// would beat every later one. settle_market has to skip it and resolve on the +// earliest capture taken at or after settle_time, and the scheduler must not count +// it as captured, or the market waits on a capture that can never settle it. +func testSettleIgnoresACaptureTakenBeforeSettleTime(t *testing.T) func(context.Context, *kwilTesting.Platform) error { + return func(ctx context.Context, platform *kwilTesting.Platform) error { + const settleTime = int64(100) + sc := newCaptureScenario(t, ctx, platform, + "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "stcapturebeforesettle00000000000", settleTime) + + sc.insert(1000, "2.000000000000000000") + sc.capture(10, settleTime-1) // one second early; sees 2 -> TRUE + + status, err := sc.ops.CaptureStatusFor(ctx, sc.marketHash, settleTime) + require.NoError(t, err) + require.Equal(t, internal.CaptureStatus{}, status, + "an early capture must not count, so the scheduler captures the market again") + + settleErr := sc.settle() + require.Error(t, settleErr, "the only capture is early, so nothing can settle the market yet") + require.Contains(t, settleErr.Error(), "taken before settle_time") + require.Nil(t, sc.winningOutcome()) + + sc.insert(1200, "0.000000000000000000") + sc.capture(20, settleTime) // exactly at settle time; sees 0 -> FALSE + + status, err = sc.ops.CaptureStatusFor(ctx, sc.marketHash, settleTime) + require.NoError(t, err) + require.Equal(t, internal.CaptureStatus{Captured: true, Signed: true, CapturedAt: 20}, status, + "a capture at settle time counts, and the early one at height 10 does not") + + require.NoError(t, sc.settle()) + winning := sc.winningOutcome() + require.NotNil(t, winning, "market should be settled") + require.False(t, *winning, + "settlement must resolve on the capture taken at settle time (0 -> FALSE), not the earlier one at height 10 (2 -> TRUE)") + + t.Logf("✅ settlement skipped the capture taken before settle_time") + return nil + } +} + +// A capture written before attestations had created_timestamp has no time, and +// its block time cannot be recovered from the table. It cannot be placed against +// settle_time, so it is treated like an early one: skipped, and the market is +// captured again. +func testSettleIgnoresACaptureWithNoTime(t *testing.T) func(context.Context, *kwilTesting.Platform) error { + return func(ctx context.Context, platform *kwilTesting.Platform) error { + const settleTime = int64(100) + sc := newCaptureScenario(t, ctx, platform, + "0xcccccccccccccccccccccccccccccccccccccccc", "stcapturewithnotime0000000000000", settleTime) + + sc.insert(1000, "2.000000000000000000") + sc.capture(10, settleTime+50) // sees 2 -> TRUE + + // The row as it reads after the column was added to a table that + // already held it. + _, err := platform.DB.Execute(ctx, + `UPDATE main.attestations SET created_timestamp = NULL WHERE attestation_hash = $1`, sc.marketHash) + require.NoError(t, err) + + status, err := sc.ops.CaptureStatusFor(ctx, sc.marketHash, settleTime) + require.NoError(t, err) + require.Equal(t, internal.CaptureStatus{}, status, "a capture with no time must not count") + + settleErr := sc.settle() + require.Error(t, settleErr) + require.Contains(t, settleErr.Error(), "no recorded time") + + sc.insert(1200, "0.000000000000000000") + sc.capture(20, settleTime+60) // sees 0 -> FALSE + + require.NoError(t, sc.settle()) + winning := sc.winningOutcome() + require.NotNil(t, winning, "market should be settled") + require.False(t, *winning, + "settlement must resolve on the timed capture (0 -> FALSE), not the one with no time (2 -> TRUE)") + + t.Logf("✅ settlement skipped the capture with no recorded time") + return nil + } +} diff --git a/internal/migrations/023-attestation-schema.sql b/internal/migrations/023-attestation-schema.sql index 6ec922430..63a0334d0 100644 --- a/internal/migrations/023-attestation-schema.sql +++ b/internal/migrations/023-attestation-schema.sql @@ -22,6 +22,18 @@ CREATE TABLE IF NOT EXISTS attestations ( CONSTRAINT chk_att_encrypt_sig_false CHECK (encrypt_sig = false) ); +-- Block time (unix seconds) of the block a capture was taken in. +-- +-- A market resolves against the value that stood at its settle_time, a unix +-- timestamp, and created_height alone cannot be compared with that. This is what +-- lets settle_market tell a capture taken before settle_time from one taken after. +-- +-- Rows written before this column existed keep it NULL: their block time is not +-- recoverable from attestations. Settlement never resolves on a capture without a +-- time, so a market whose only captures predate this column is captured again. +ALTER TABLE attestations +ADD COLUMN IF NOT EXISTS created_timestamp INT8; + -- Allowlist table for actions permitted for attestation CREATE TABLE IF NOT EXISTS attestation_actions ( action_name TEXT PRIMARY KEY, diff --git a/internal/migrations/024-attestation-actions.prod.sql b/internal/migrations/024-attestation-actions.prod.sql index 5b5c60fa1..f8c198797 100644 --- a/internal/migrations/024-attestation-actions.prod.sql +++ b/internal/migrations/024-attestation-actions.prod.sql @@ -74,9 +74,11 @@ CREATE OR REPLACE ACTION request_attestation( eth_truf.transfer($leader_addr, $attestation_fee); -- ===== END FEE COLLECTION ===== - -- Get current block height + -- Get current block height, and the block time settlement compares with a + -- market's settle_time $created_height := @height; - + $created_timestamp := @block_timestamp; + -- Normalize caller address to bytes for storage (re-use safe normalization) $caller_bytes := $caller_bytes; -- Already normalized above @@ -150,10 +152,12 @@ CREATE OR REPLACE ACTION request_attestation( -- Store unsigned attestation INSERT INTO attestations ( request_tx_id, attestation_hash, requester, data_provider, stream_id, - result_canonical, encrypt_sig, created_height, signature, validator_pubkey, signed_height + result_canonical, encrypt_sig, created_height, created_timestamp, + signature, validator_pubkey, signed_height ) VALUES ( $request_tx_id, $attestation_hash, $caller_bytes, $data_provider, $stream_id, - $result_canonical, $encrypt_sig, $created_height, NULL, NULL, NULL + $result_canonical, $encrypt_sig, $created_height, $created_timestamp, + NULL, NULL, NULL ); -- Queue for signing (no-op on non-leader validators; handled by precompile) diff --git a/internal/migrations/024-attestation-actions.sql b/internal/migrations/024-attestation-actions.sql index eab44e22f..ad5943d20 100644 --- a/internal/migrations/024-attestation-actions.sql +++ b/internal/migrations/024-attestation-actions.sql @@ -70,9 +70,11 @@ CREATE OR REPLACE ACTION request_attestation( hoodi_tt.transfer($leader_addr, $attestation_fee); -- ===== END FEE COLLECTION ===== - -- Get current block height + -- Get current block height, and the block time settlement compares with a + -- market's settle_time $created_height := @height; - + $created_timestamp := @block_timestamp; + -- Normalize caller address to bytes for storage (re-use safe normalization) $caller_bytes := $caller_bytes; -- Already normalized above @@ -146,10 +148,12 @@ CREATE OR REPLACE ACTION request_attestation( -- Store unsigned attestation INSERT INTO attestations ( request_tx_id, attestation_hash, requester, data_provider, stream_id, - result_canonical, encrypt_sig, created_height, signature, validator_pubkey, signed_height + result_canonical, encrypt_sig, created_height, created_timestamp, + signature, validator_pubkey, signed_height ) VALUES ( $request_tx_id, $attestation_hash, $caller_bytes, $data_provider, $stream_id, - $result_canonical, $encrypt_sig, $created_height, NULL, NULL, NULL + $result_canonical, $encrypt_sig, $created_height, $created_timestamp, + NULL, NULL, NULL ); -- Queue for signing (no-op on non-leader validators; handled by precompile) diff --git a/internal/migrations/032-order-book-actions.sql b/internal/migrations/032-order-book-actions.sql index 5850f3862..d292bb5be 100644 --- a/internal/migrations/032-order-book-actions.sql +++ b/internal/migrations/032-order-book-actions.sql @@ -2481,7 +2481,8 @@ CREATE OR REPLACE ACTION change_ask( * * Settlement Process: * 1. Validate market exists, not already settled, and settle_time reached - * 2. Query attestation by market hash (market.hash = attestation.attestation_hash) + * 2. Query attestation by market hash (market.hash = attestation.attestation_hash), + * taken at or after settle_time * 3. Verify attestation has been signed * 4. Parse result_canonical to extract boolean outcome (TRUE = YES wins, FALSE = NO wins) * 5. Mark market as settled with winning_outcome and settled_at timestamp @@ -2501,6 +2502,7 @@ CREATE OR REPLACE ACTION change_ask( * - If market is already settled * - If settle_time has not been reached * - If attestation doesn't exist for market hash + * - If every attestation for market hash was taken before settle_time * - If attestation is not yet signed * - If result parsing fails * @@ -2622,11 +2624,20 @@ CREATE OR REPLACE ACTION settle_market( -- asks for another. The two captures read the query against different chain -- state and are not interchangeable. -- - -- Resolve on the EARLIEST signed capture: it is the one taken closest to - -- settle_time, which is the state the market was defined to resolve against. - -- Ordering by signed_height instead made the outcome a function of when a - -- validator got to each row, and when two were signed in the same block that - -- ordering was a tie decided by whichever row the database returned first. + -- Resolve on the EARLIEST signed capture taken at or after settle_time: it is + -- the one taken closest to settle_time, which is the state the market was + -- defined to resolve against. Ordering by signed_height instead made the + -- outcome a function of when a validator got to each row, and when two were + -- signed in the same block that ordering was a tie decided by whichever row + -- the database returned first. + -- + -- A capture taken before settle_time never resolves the market. + -- request_attestation is PUBLIC and a live market's query_components are + -- readable, so anyone can capture its query while the window it settles on + -- is still open, and the earliest-first ordering would then prefer that + -- capture to every later one. A capture with no created_timestamp was taken + -- before the column existed and cannot be placed against settle_time, so it + -- is skipped as well (NULL >= settle_time is not true). $result_canonical BYTEA; $capture_height INT8; $attestation_found BOOL := false; @@ -2635,6 +2646,7 @@ CREATE OR REPLACE ACTION settle_market( FROM attestations WHERE attestation_hash = $market_hash AND signature IS NOT NULL + AND created_timestamp >= $settle_time ORDER BY created_height ASC LIMIT 1 { $result_canonical := $row.result_canonical; @@ -2643,9 +2655,22 @@ CREATE OR REPLACE ACTION settle_market( } if NOT $attestation_found { - -- Nothing signed. Say which of the two reasons it is, because they need - -- different responses: a capture that exists is waiting on a validator, - -- and one that does not has to be requested. + -- Nothing usable is signed. Say which reason it is, because they need + -- different responses: a capture taken in time is waiting on a + -- validator, and otherwise a new capture has to be requested. + $capture_in_time BOOL := false; + for $row in SELECT 1 AS present + FROM attestations + WHERE attestation_hash = $market_hash + AND created_timestamp >= $settle_time + LIMIT 1 { + $capture_in_time := true; + } + + if $capture_in_time { + ERROR('Attestation not yet signed by validator. Please wait for signing to complete.'); + } + $capture_exists BOOL := false; for $row in SELECT 1 AS present FROM attestations @@ -2655,7 +2680,8 @@ CREATE OR REPLACE ACTION settle_market( } if $capture_exists { - ERROR('Attestation not yet signed by validator. Please wait for signing to complete.'); + ERROR('Every attestation for market hash was taken before settle_time ' || $settle_time::TEXT || + ' or has no recorded time. Request a new attestation.'); } ERROR('Attestation not found for market hash. Market cannot be settled without attestation.'); } diff --git a/tests/streams/order_book/order_events_test.go b/tests/streams/order_book/order_events_test.go index 95757f276..567926176 100644 --- a/tests/streams/order_book/order_events_test.go +++ b/tests/streams/order_book/order_events_test.go @@ -621,8 +621,13 @@ func testOrderEventSettlement(t *testing.T) func(ctx context.Context, platform * }) require.NoError(t, err) + // The market below settles an hour from now, and settle_market only takes + // a capture taken from its settle time on. + settleTime := time.Now().Add(1 * time.Hour).Unix() + var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = settleTime res, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -638,7 +643,6 @@ func testOrderEventSettlement(t *testing.T) func(ctx context.Context, platform * queryComponents, err := encodeQueryComponentsForTests(dataProvider, streamID, "get_last_record", argsBytes) require.NoError(t, err) - settleTime := time.Now().Add(1 * time.Hour).Unix() var queryID int engineCtx = helper.NewEngineContext() diff --git a/tests/streams/order_book/settlement_payout_test.go b/tests/streams/order_book/settlement_payout_test.go index f7522abed..5fef75501 100644 --- a/tests/streams/order_book/settlement_payout_test.go +++ b/tests/streams/order_book/settlement_payout_test.go @@ -170,6 +170,11 @@ func testWinnerReceivesFullPayout(t *testing.T) func(context.Context, *kwilTesti }) require.NoError(t, err) + // The market below settles an hour from now, and settle_market only takes + // a capture taken from its settle time on. + settleTime := time.Now().Add(1 * time.Hour).Unix() + dpEngineCtx.TxContext.BlockContext.Timestamp = settleTime + var requestTxID string var attestationHash []byte // Attestations are typically requested by users, but can be anyone @@ -205,7 +210,6 @@ func testWinnerReceivesFullPayout(t *testing.T) func(context.Context, *kwilTesti require.NoError(t, err) // Create market using query_components - settleTime := time.Now().Add(1 * time.Hour).Unix() // Future timestamp maxSpread := int64(5) minOrderSize := int64(1) var queryID int diff --git a/tests/streams/order_book/settlement_test.go b/tests/streams/order_book/settlement_test.go index 0fe0d3b08..a22cf697b 100644 --- a/tests/streams/order_book/settlement_test.go +++ b/tests/streams/order_book/settlement_test.go @@ -21,6 +21,11 @@ import ( attestationTests "github.com/trufnetwork/node/tests/streams/attestation" ) +// captureTime is the block time this file's captures are taken at for markets +// that settle at 100: settle_market resolves a market only on a capture taken at +// or after its settle time. +const captureTime = int64(150) + // NO_OUTCOME_VALUE represents a NO outcome in settlement tests. // We use -1.0 instead of 0.0 because insert_records filters WHERE value != 0, // which prevents inserting exact 0.0 values. The parse_attestation_boolean @@ -145,6 +150,8 @@ func testSettleMarketHappyPath(t *testing.T) func(context.Context, *kwilTesting. var requestTxID string engineCtx = helper.NewEngineContext() + settleTime := time.Now().Add(1 * time.Hour).Unix() + engineCtx.TxContext.BlockContext.Timestamp = settleTime // settle_market only takes a capture from settle_time on res, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{ dataProvider, @@ -176,7 +183,6 @@ func testSettleMarketHappyPath(t *testing.T) func(context.Context, *kwilTesting. require.NoError(t, err) // Create market using query_components - settleTime := time.Now().Add(1 * time.Hour).Unix() maxSpread := int64(5) minOrderSize := int64(1) var queryID int @@ -308,6 +314,7 @@ func testSettleMarketWithNoOutcome(t *testing.T) func(context.Context, *kwilTest var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -419,6 +426,7 @@ func testSettleMarketWithMultipleDatapoints(t *testing.T) func(context.Context, var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -545,6 +553,7 @@ func testSettleMarketAlreadySettled(t *testing.T) func(context.Context, *kwilTes var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -640,6 +649,7 @@ func testSettleMarketTooEarly(t *testing.T) func(context.Context, *kwilTesting.P var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = 1000 // the market's settle time below _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -779,6 +789,7 @@ func testSettleMarketAttestationNotSigned(t *testing.T) func(context.Context, *k }) engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -881,6 +892,7 @@ func testSettleMarketValidationIntegration(t *testing.T) func(context.Context, * var requestTxID string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -989,6 +1001,8 @@ func testSettleMarketBlockedByBinaryParityViolation(t *testing.T) func(context.C var requestTxID string engineCtx = helper.NewEngineContext() + settleTime := time.Now().Add(1 * time.Hour).Unix() + engineCtx.TxContext.BlockContext.Timestamp = settleTime // settle_market only takes a capture from settle_time on _, err = platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID, "get_last_record", argsBytes, false, nil}, func(row *common.Row) error { @@ -1004,7 +1018,6 @@ func testSettleMarketBlockedByBinaryParityViolation(t *testing.T) func(context.C require.NoError(t, err) // Create market - settleTime := time.Now().Add(1 * time.Hour).Unix() var queryID int engineCtx = helper.NewEngineContext() engineCtx.TxContext.BlockContext.Timestamp = time.Now().Unix() @@ -1149,6 +1162,7 @@ func testSettleMarketMultiMarketCollateral(t *testing.T) func(context.Context, * require.NoError(t, err) var requestTxID1 string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime res1, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID1, "get_last_record", argsBytes1, false, nil}, func(row *common.Row) error { @@ -1191,6 +1205,7 @@ func testSettleMarketMultiMarketCollateral(t *testing.T) func(context.Context, * require.NoError(t, err) var requestTxID2 string engineCtx = helper.NewEngineContext() + engineCtx.TxContext.BlockContext.Timestamp = captureTime res2, err := platform.Engine.Call(engineCtx, platform.DB, "", "request_attestation", []any{dataProvider, streamID2, "get_last_record", argsBytes2, false, nil}, func(row *common.Row) error { From 12f094ba205f3a3230841db67db7b296998cfe3b Mon Sep 17 00:00:00 2001 From: Michael Buntarman Date: Thu, 8 Oct 2026 16:56:09 +0700 Subject: [PATCH 2/2] chore: ignore captures taken before a market's settle time --- extensions/tn_settlement/internal/engine_ops.go | 2 ++ extensions/tn_settlement/settlement_integration_test.go | 1 + 2 files changed, 3 insertions(+) diff --git a/extensions/tn_settlement/internal/engine_ops.go b/extensions/tn_settlement/internal/engine_ops.go index 385c19564..b85146184 100644 --- a/extensions/tn_settlement/internal/engine_ops.go +++ b/extensions/tn_settlement/internal/engine_ops.go @@ -382,6 +382,8 @@ func (e *EngineOperations) AttestationExists(ctx context.Context, marketHash []b // 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 diff --git a/extensions/tn_settlement/settlement_integration_test.go b/extensions/tn_settlement/settlement_integration_test.go index eb1def081..48f627635 100644 --- a/extensions/tn_settlement/settlement_integration_test.go +++ b/extensions/tn_settlement/settlement_integration_test.go @@ -1061,6 +1061,7 @@ func (sc *captureScenario) settle() error { return res.Error } +// winningOutcome reads the market's winning outcome, or nil while it is unsettled. func (sc *captureScenario) winningOutcome() *bool { var winning *bool require.NoError(sc.t, sc.platform.Engine.Execute(sc.helper.NewEngineContext(), sc.platform.DB,