Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/build-cnpg-timescaledb-dev.yml
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@ on:
default: '17'
required: true
tsdb_version:
description: TimescaleDB version (semver, e.g. 2.25.0)
default: '2.25.0'
description: TimescaleDB version (semver, e.g. 2.28.2)
default: '2.28.2'
required: true

permissions:
Expand Down
4 changes: 2 additions & 2 deletions .github/workflows/build-cnpg-timescaledb-stg.yml
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@ on:
default: '17'
required: true
tsdb_version:
description: TimescaleDB version (semver, e.g. 2.25.0)
default: '2.25.0'
description: TimescaleDB version (semver, e.g. 2.28.2)
default: '2.28.2'
required: true

permissions:
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/go.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ jobs:
runs-on: ubuntu-latest
services:
postgres:
image: timescale/timescaledb:2.25.0-pg17
image: timescale/timescaledb:2.28.2-pg17
env:
POSTGRES_USER: postgres
POSTGRES_DB: postgres
Expand Down
4 changes: 2 additions & 2 deletions Dockerfile-timescale-cnpg
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
# TimescaleDB from the apt repo and layers it onto the CNPG base image.
#
# Usage:
# docker build -f Dockerfile-timescale-cnpg --build-arg PG_MAJOR=17 --build-arg TSDB_VERSION=2.25.0 -t cnpg-timescaledb .
# docker build -f Dockerfile-timescale-cnpg --build-arg PG_MAJOR=17 --build-arg TSDB_VERSION=2.28.2 -t cnpg-timescaledb .

ARG PG_MAJOR=17

Expand All @@ -15,7 +15,7 @@ ARG PG_MAJOR=17
FROM ghcr.io/cloudnative-pg/postgresql:${PG_MAJOR}-bookworm AS timescaledb-builder

ARG PG_MAJOR
ARG TSDB_VERSION=2.25.0
ARG TSDB_VERSION=2.28.2

USER root

Expand Down
2 changes: 1 addition & 1 deletion docker-compose.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ services:

db:
container_name: db
image: timescale/timescaledb:2.25.0-pg17
image: timescale/timescaledb:2.28.2-pg17
command: ["postgres", "-c", "timescaledb.enable_chunk_skipping=on", "-c", "timescaledb.enable_sparse_index_bloom=on"]
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres -d wallet-backend"]
Expand Down
19 changes: 15 additions & 4 deletions docs/data-migrations/adding-a-protocol.md
Original file line number Diff line number Diff line change
Expand Up @@ -166,13 +166,12 @@ CREATE TABLE <protocol>_<entity> (
last_modified_ledger INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (...) -- Choose a primary key that fits your state's natural identity
) WITH (
fillfactor = 80,
fillfactor = 90,
autovacuum_vacuum_scale_factor = 0.02,
autovacuum_vacuum_threshold = 50,
autovacuum_analyze_scale_factor = 0.01,
autovacuum_analyze_threshold = 50,
autovacuum_vacuum_cost_delay = 0,
autovacuum_vacuum_cost_limit = 1000
autovacuum_vacuum_cost_delay = 0
);

-- Add indexes based on your query patterns
Expand All @@ -182,7 +181,19 @@ CREATE TABLE <protocol>_<entity> (
DROP TABLE IF EXISTS <protocol>_<entity>;
```

The storage parameters (`fillfactor = 80`, aggressive autovacuum) are tuned for heavy UPSERT workloads during ingestion. Adjust based on your protocol's write pattern.
`fillfactor = 90` is the default: it reserves 10% free space per page for HOT (Heap-Only
Tuple) updates without over-reserving for tables whose UPSERTs only touch non-indexed
columns. Use `fillfactor = 80` instead only for tables whose rows update on essentially
every ledger (e.g. AMM-style reserves that shift on every trade). `autovacuum_vacuum_cost_delay
= 0` disables cost-based vacuum throttling for this table's worker; there is no
`autovacuum_vacuum_cost_limit` to set alongside it since a delay of 0 makes any cost
limit a no-op.

The aggressive autovacuum block above is for tables whose row count grows with users or
usage (positions, balances, allowances). Skip it entirely — keep only `fillfactor = 90`
— for small, protocol-bounded current-state tables (pool configs, reserve configs) whose
row count is capped by the number of pools/reserves your protocol will ever have
deployed; default autovacuum behavior is fine for those.

### Data Model

Expand Down
14 changes: 14 additions & 0 deletions docs/operations.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
# Operations

**Hypertable index usage:** `pg_stat_user_indexes` on the `public` schema always reports `idx_scan = 0` for hypertable indexes — scans accrue on the per-chunk indexes in `_timescaledb_internal`, and the parent's index entry is only a template. To measure real usage (e.g. for an unused-index audit), aggregate chunk-index stats:

```sql
SELECT ht.table_name AS hypertable, regexp_replace(sui.indexrelname, '^_hyper_\d+_\d+_chunk_', '') AS index_name,
sum(sui.idx_scan) AS scans
FROM pg_stat_user_indexes sui
JOIN _timescaledb_catalog.chunk c ON c.schema_name = sui.schemaname AND c.table_name = sui.relname
JOIN _timescaledb_catalog.hypertable ht ON ht.id = c.hypertable_id
GROUP BY 1, 2 ORDER BY 1, 3 DESC;
```

Every index in the schema must have a nameable consumer query; secondary indexes duplicating a primary key's column set are not kept (the PK's column order is chosen to serve its consumers directly — B-tree scans run forward or backward, so ASC/DESC variants of the same key are one index, not two).
24 changes: 20 additions & 4 deletions internal/data/ingest_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -169,19 +169,35 @@ func (m *IngestStoreModel) UpdateMin(ctx context.Context, dbTx pgx.Tx, cursorNam
return nil
}

func (m *IngestStoreModel) GetLedgerGaps(ctx context.Context) ([]LedgerRange, error) {
// GetLedgerGaps returns gaps between consecutive present ledger_number values within
// [startLedger, endLedger]. The window bounds the cost two ways: chunks whose recorded
// ledger_number range (enable_chunk_skipping on transactions.ledger_number; ranges are
// captured when a chunk is compressed) doesn't overlap the window are excluded at plan
// time, and any remaining out-of-window batches are dropped by the columnstore's batch
// min/max metadata without decompression.
//
// Edge semantics: callers must pass a startLedger that is itself a present ledger_number (e.g.
// the oldest ingested ledger) — the left edge is NOT synthesized. If startLedger fell strictly
// inside an already-open gap, this function would not report the portion before the first
// present row in-window; this mirrors the original unwindowed behavior, which never reported a
// gap before the very first row in the whole table. The right edge IS handled: COALESCE falls
// back to endLedger+1 when LEAD finds no next row within the window (the next present ledger
// lies beyond endLedger, or doesn't exist yet), so a gap still open at the window's boundary is
// reported clipped to endLedger instead of silently dropped (plain LEAD would return NULL there,
// failing the gap_start <= gap_end filter and losing the trailing partial gap).
func (m *IngestStoreModel) GetLedgerGaps(ctx context.Context, startLedger, endLedger uint32) ([]LedgerRange, error) {
const query = `
SELECT gap_start, gap_end FROM (
SELECT
ledger_number + 1 AS gap_start,
LEAD(ledger_number) OVER (ORDER BY ledger_number) - 1 AS gap_end
FROM (SELECT DISTINCT ledger_number FROM transactions) t
COALESCE(LEAD(ledger_number) OVER (ORDER BY ledger_number), $2 + 1) - 1 AS gap_end
FROM (SELECT DISTINCT ledger_number FROM transactions WHERE ledger_number BETWEEN $1 AND $2) t
) gaps
WHERE gap_start <= gap_end
ORDER BY gap_start
`
start := time.Now()
ledgerGaps, err := db.QueryMany[LedgerRange](ctx, m.DB, query)
ledgerGaps, err := db.QueryMany[LedgerRange](ctx, m.DB, query, int32(startLedger), int32(endLedger))
duration := time.Since(start).Seconds()
m.Metrics.QueryDuration.WithLabelValues("GetLedgerGaps", "transactions").Observe(duration)
m.Metrics.QueriesTotal.WithLabelValues("GetLedgerGaps", "transactions").Inc()
Expand Down
62 changes: 57 additions & 5 deletions internal/data/ingest_store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -401,15 +401,21 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {

testCases := []struct {
name string
startLedger uint32
endLedger uint32
setupDB func(t *testing.T)
expectedGaps []LedgerRange
}{
{
name: "returns_empty_when_no_transactions",
startLedger: 1,
endLedger: 100,
expectedGaps: []LedgerRange{},
},
{
name: "returns_empty_when_no_gaps",
name: "returns_empty_when_no_gaps",
startLedger: 100,
endLedger: 102,
setupDB: func(t *testing.T) {
// Insert consecutive ledgers: 100, 101, 102
for i, ledger := range []uint32{100, 101, 102} {
Expand All @@ -423,7 +429,9 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
expectedGaps: []LedgerRange{},
},
{
name: "returns_single_gap",
name: "returns_single_gap",
startLedger: 100,
endLedger: 105,
setupDB: func(t *testing.T) {
// Insert ledgers 100 and 105, creating gap 101-104
for i, ledger := range []uint32{100, 105} {
Expand All @@ -439,7 +447,9 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
},
},
{
name: "returns_multiple_gaps",
name: "returns_multiple_gaps",
startLedger: 100,
endLedger: 110,
setupDB: func(t *testing.T) {
// Insert ledgers 100, 105, 110, creating gaps 101-104 and 106-109
for i, ledger := range []uint32{100, 105, 110} {
Expand All @@ -456,7 +466,9 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
},
},
{
name: "handles_single_ledger_gap",
name: "handles_single_ledger_gap",
startLedger: 100,
endLedger: 102,
setupDB: func(t *testing.T) {
// Insert ledgers 100 and 102, creating gap of just 101
for i, ledger := range []uint32{100, 102} {
Expand All @@ -471,6 +483,46 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
{GapStart: 101, GapEnd: 101},
},
},
{
// SQL-05: the real next ledger (200) lies beyond the requested window (150). The
// trailing gap must be reported clipped to endLedger, not dropped just because LEAD
// finds no next row inside the window.
name: "clips_trailing_gap_still_open_at_window_boundary",
startLedger: 100,
endLedger: 150,
setupDB: func(t *testing.T) {
for i, ledger := range []uint32{100, 200} {
_, err := dbConnectionPool.Exec(ctx,
`INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at)
VALUES ($1, $2, 100, 'TransactionResultCodeTxSuccess', $3, NOW())`,
fmt.Sprintf("hash%d", i), i+1, ledger)
require.NoError(t, err)
}
},
expectedGaps: []LedgerRange{
{GapStart: 101, GapEnd: 150},
},
},
{
// Ledgers outside [startLedger, endLedger] must not affect the result: a gap beyond
// endLedger (106-109) is excluded entirely, proving the query is bounded rather than
// scanning the whole table.
name: "ignores_gaps_entirely_outside_the_window",
startLedger: 100,
endLedger: 104,
setupDB: func(t *testing.T) {
for i, ledger := range []uint32{100, 105, 110} {
_, err := dbConnectionPool.Exec(ctx,
`INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at)
VALUES ($1, $2, 100, 'TransactionResultCodeTxSuccess', $3, NOW())`,
fmt.Sprintf("hash%d", i), i+1, ledger)
require.NoError(t, err)
}
},
expectedGaps: []LedgerRange{
{GapStart: 101, GapEnd: 104},
},
},
}

for _, tc := range testCases {
Expand All @@ -490,7 +542,7 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
tc.setupDB(t)
}

gaps, err := m.GetLedgerGaps(ctx)
gaps, err := m.GetLedgerGaps(ctx, tc.startLedger, tc.endLedger)
require.NoError(t, err)
assert.Equal(t, tc.expectedGaps, gaps)
})
Expand Down
8 changes: 0 additions & 8 deletions internal/data/mocks.go
Original file line number Diff line number Diff line change
Expand Up @@ -309,14 +309,6 @@ func (m *ProtocolsModelMock) GetByIDs(ctx context.Context, protocolIDs []string)
return args.Get(0).([]Protocols), args.Error(1)
}

func (m *ProtocolsModelMock) GetClassified(ctx context.Context) ([]Protocols, error) {
args := m.Called(ctx)
if args.Get(0) == nil {
return nil, args.Error(1)
}
return args.Get(0).([]Protocols), args.Error(1)
}

func (m *ProtocolsModelMock) InsertIfNotExists(ctx context.Context, dbTx pgx.Tx, protocolID string) error {
args := m.Called(ctx, dbTx, protocolID)
return args.Error(0)
Expand Down
Loading
Loading