Pre-prod hardening 4/4: GraphQL API hardening + ops docs

cmd/serve.go

@@ -29,6 +29,7 @@ func (c *serveCmd) Command() *cobra.Command {
 		utils.StellarEnvironmentOption(&stellarEnvironment),
 		utils.ServerBaseURLOption(&cfg.ServerBaseURL),
 		utils.GraphQLComplexityLimitOption(&cfg.GraphQLComplexityLimit),
+		utils.GraphQLIntrospectionEnabledOption(&cfg.GraphQLIntrospectionEnabled),
 		utils.AdminPortOption(&cfg.AdminPort),
 		{
 			Name:        "port",

cmd/utils/global_options.go

@@ -164,6 +164,20 @@ func GraphQLComplexityLimitOption(configKey *int) *config.ConfigOption {
 	}
 }
 
+// GraphQLIntrospectionEnabledOption controls whether GraphQL schema introspection (__schema,
+// __type) is served. Defaults to disabled: introspection makes the full schema (including any
+// unreleased or internal-only fields) discoverable to anyone who can reach the endpoint.
+func GraphQLIntrospectionEnabledOption(configKey *bool) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "graphql-introspection-enabled",
+		Usage:       "Whether to enable GraphQL schema introspection (__schema, __type). Disabled by default in production.",
+		OptType:     types.Bool,
+		ConfigKey:   configKey,
+		FlagDefault: false,
+		Required:    false,
+	}
+}
+
 // DBPoolOptions returns config options for tuning the pgxpool connection pool.
 func DBPoolOptions(maxConns *int, minConns *int, maxConnLifetime *time.Duration, maxConnIdleTime *time.Duration) config.ConfigOptions {
 	return config.ConfigOptions{

docs/operations.md

@@ -1,4 +1,48 @@
-# Operations
+# Production Operations
+
+Reference for operating the wallet-backend's TimescaleDB-backed ingestion and API in production.
+
+## Production Database Checklist
+
+**CNPG `postgresql.parameters`:**
+
+| Parameter | Value | Why |
+|---|---|---|
+| `timescaledb.enable_chunk_skipping` | `on` | Off by default. Min/max ranges used for chunk skipping are recorded at compression time, so this must be on **before** the first ingest — enabling it later requires recompressing all existing history to backfill the ranges. |
+| `timescaledb.enable_sparse_index_bloom` | `on` | Enables bloom-filter sparse indexes on compressed chunks. |
+| `timescaledb.stats_max_chunks` | `2048` (2.28+ only) | Steady-state compressed chunk count at 1-year retention is ~1,820, which exceeds the 1,024 default stats cache. |
+
+**Instance sizing:**
+
+| Resource | Target | Why |
+|---|---|---|
+| Memory | >= 64 GB, `shared_buffers` 16 GB, `effective_cache_size` ~48 GB | The actively-ingested day's chunk indexes total ~14 GB, fitting the shared_buffers 25%-of-RAM guidance. |
+| Volume | >= 1.5 TB | Steady state at 1-year retention is ~1 TB (roughly 2.3 GB/day compressed across hypertables, plus balance tables), with the remainder as WAL/headroom. |
+
+**Runtime flags** (`ingest` command):
+
+| Flag / env var | Value | Why |
+|---|---|---|
+| `RETENTION_PERIOD` | `1 year` | Chunks older than this are dropped automatically. |
+| `COMPRESSION_COMPRESS_AFTER` | `1 day` | Leaves one uncompressed day of headroom so gap-backfill writes land in the rowstore instead of a compressed chunk. |
+| `COMPRESSION_SCHEDULE_INTERVAL` | `1 day` | How often the compression job checks for eligible chunks. |
+| `COMPRESSION_MAX_CHUNKS` | `10` | Caps chunks compressed per job run to prevent overlapping runs. |
+| `CHUNK_INTERVAL` | `1 day` | Hypertable chunk time interval; only affects newly created chunks. |
+
+**GraphQL:** `GRAPHQL_INTROSPECTION_ENABLED` should be unset (or `false`) in production. Introspection makes the full schema, including any unreleased or internal-only fields, discoverable to anyone who can reach the endpoint. Leave it `true` in dev environments.
+
+## Alerting & Failure Modes
+
+**Alert on:**
+- `wallet_ingestion_lag_ledgers` growth — ingestion falling behind the backend tip.
+- Container restart counts — the ingest process exits immediately after a permanent failure.
+- `wallet_db_query_errors_total{error_type="cursor_missing"}` — fires only when a protocol cursor that previously existed disappears, which is always a genuine incident (not a normal not-yet-initialized state).
+
+**Do not alert on** `wallet_ingestion_retry_exhaustions_total` directly — the process exits right after incrementing it, so by the time an alert fires the process has already restarted. Treat it as a post-mortem breadcrumb, not a live signal.
+
+**Poisoned ledgers:** a ledger that deterministically fails ingestion cannot self-heal by retrying — the ingest loop fails fast on permanent SQLSTATE classes, and recovery requires either a code fix or a manually-guarded cursor bump past the ledger.
+
+**TimescaleDB job health** has no application-level metric. Monitor `timescaledb_information.job_stats` and `timescaledb_information.job_errors` directly via SQL (through the CNPG metrics exporter or a scheduled check) to catch stalled or failing compression/retention jobs.
 
 **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:
 
@@ -12,3 +56,7 @@ 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).
+
+## Connection Topology Constraint
+
+Ingestion connects directly to the CNPG `rw` service and relies on pgx's server-side prepared-statement caching for performance. If a transaction-pooling PgBouncer/Pooler is ever placed in front of the ingest path, its pool mode must be switched to `QueryExecModeExec` — transaction pooling is incompatible with server-side prepared statements. The API server already runs in `Exec` mode.

internal/data/ingest_store.go

@@ -198,7 +198,7 @@ func (m *IngestStoreModel) CompareAndSwap(ctx context.Context, dbTx pgx.Tx, curs
 func (m *IngestStoreModel) UpdateMin(ctx context.Context, dbTx pgx.Tx, cursorName string, ledger uint32) error {
 	const query = `
 		UPDATE ingest_store
-		SET value = LEAST(value::integer, $2)::text
+		SET value = LEAST(value::bigint, $2)::text
 		WHERE key = $1
 	`
 	start := time.Now()

internal/data/operations.go

@@ -37,56 +37,6 @@ func (m *OperationModel) GetByID(ctx context.Context, id int64, columns string)
 	return &operation, nil
 }
 
-func (m *OperationModel) GetAll(ctx context.Context, columns string, limit *int32, cursor *types.CompositeCursor, sortOrder SortOrder) ([]*types.OperationWithCursor, error) {
-	if err := validatePositiveLimit(limit); err != nil {
-		return nil, err
-	}
-	columns = prepareColumnsWithID(columns, types.Operation{}, "", "id", "ledger_created_at")
-	queryBuilder := strings.Builder{}
-	var args []interface{}
-	argIndex := 1
-
-	fmt.Fprintf(&queryBuilder, `SELECT %s, ledger_created_at as cursor_ledger_created_at, id as cursor_id FROM operations`, columns)
-
-	// Decomposed cursor pagination: expands ROW() tuple comparison into OR clauses so
-	// TimescaleDB ColumnarScan can push filters into vectorized batch processing.
-	if cursor != nil {
-		clause, cursorArgs, nextIdx := buildDecomposedCursorCondition([]CursorColumn{
-			{Name: "ledger_created_at", Value: cursor.LedgerCreatedAt},
-			{Name: "id", Value: cursor.ID},
-		}, sortOrder, argIndex)
-		queryBuilder.WriteString(" WHERE " + clause)
-		args = append(args, cursorArgs...)
-		argIndex = nextIdx
-	}
-
-	if sortOrder == DESC {
-		queryBuilder.WriteString(" ORDER BY ledger_created_at DESC, id DESC")
-	} else {
-		queryBuilder.WriteString(" ORDER BY ledger_created_at ASC, id ASC")
-	}
-
-	if limit != nil {
-		fmt.Fprintf(&queryBuilder, " LIMIT $%d", argIndex)
-		args = append(args, *limit)
-	}
-	query := queryBuilder.String()
-	if sortOrder == DESC {
-		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY operations.cursor_ledger_created_at ASC, operations.cursor_id ASC`, query)
-	}
-
-	start := time.Now()
-	operations, err := db.QueryManyPtrs[types.OperationWithCursor](ctx, m.DB, query, args...)
-	duration := time.Since(start).Seconds()
-	m.Metrics.QueryDuration.WithLabelValues("GetAll", "operations").Observe(duration)
-	m.Metrics.QueriesTotal.WithLabelValues("GetAll", "operations").Inc()
-	if err != nil {
-		m.Metrics.QueryErrors.WithLabelValues("GetAll", "operations", utils.GetDBErrorType(err)).Inc()
-		return nil, fmt.Errorf("getting operations: %w", err)
-	}
-	return operations, nil
-}
-
 // BatchGetByToIDs gets the operations that are associated with the given transaction ToIDs.
 //
 // # TOID (Total Order ID) Encoding - SEP-35

internal/data/operations_test.go

@@ -191,67 +191,6 @@ func Test_OperationModel_BatchCopy(t *testing.T) {
 	}
 }
 
-func TestOperationModel_GetAll(t *testing.T) {
-	dbt := dbtest.Open(t)
-	defer dbt.Close()
-	ctx := context.Background()
-	dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN)
-	require.NoError(t, err)
-	defer dbConnectionPool.Close()
-
-	reg := prometheus.NewRegistry()
-	dbMetrics := metrics.NewMetrics(reg).DB
-
-	m := &OperationModel{
-		DB:      dbConnectionPool,
-		Metrics: dbMetrics,
-	}
-
-	now := time.Now()
-
-	// Create test transactions first (hash is BYTEA, using valid 64-char hex strings)
-	testHash1 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000001")
-	testHash2 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000002")
-	testHash3 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000003")
-	_, err = dbConnectionPool.Exec(ctx, `
-		INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at, is_fee_bump)
-		VALUES
-			($2, 1, 100, 'TransactionResultCodeTxSuccess', 1, $1, false),
-			($3, 2, 200, 'TransactionResultCodeTxSuccess', 2, $1, true),
-			($4, 3, 300, 'TransactionResultCodeTxSuccess', 3, $1, false)
-	`, now, testHash1, testHash2, testHash3)
-	require.NoError(t, err)
-
-	// Create test operations (IDs must be in TOID range for each transaction: (to_id, to_id + 4096))
-	xdr1 := types.XDRBytea([]byte("xdr1"))
-	xdr2 := types.XDRBytea([]byte("xdr2"))
-	xdr3 := types.XDRBytea([]byte("xdr3"))
-	_, err = dbConnectionPool.Exec(ctx, `
-		INSERT INTO operations (id, operation_type, operation_xdr, result_code, successful, ledger_number, ledger_created_at)
-		VALUES
-			(2, 'PAYMENT', $2, 'op_success', true, 1, $1),
-			(4098, 'CREATE_ACCOUNT', $3, 'op_success', true, 2, $1),
-			(8194, 'PAYMENT', $4, 'op_success', true, 3, $1)
-	`, now, xdr1, xdr2, xdr3)
-	require.NoError(t, err)
-
-	// Test GetAll without limit (gets all operations)
-	operations, err := m.GetAll(ctx, "", nil, nil, ASC)
-	require.NoError(t, err)
-	assert.Len(t, operations, 3)
-	assert.Equal(t, int64(2), operations[0].CompositeCursor.ID)
-	assert.Equal(t, int64(4098), operations[1].CompositeCursor.ID)
-	assert.Equal(t, int64(8194), operations[2].CompositeCursor.ID)
-
-	// Test GetAll with smaller limit
-	limit := int32(2)
-	operations, err = m.GetAll(ctx, "", &limit, nil, ASC)
-	require.NoError(t, err)
-	assert.Len(t, operations, 2)
-	assert.Equal(t, int64(2), operations[0].CompositeCursor.ID)
-	assert.Equal(t, int64(4098), operations[1].CompositeCursor.ID)
-}
-
 func TestOperationModel_BatchGetByToIDs(t *testing.T) {
 	dbt := dbtest.Open(t)
 	defer dbt.Close()
@@ -930,10 +869,6 @@ func TestOperationModel_MinimalProjectionHydratesLedgerCreatedAt(t *testing.T) {
 	assert.True(t, now.Equal(op.LedgerCreatedAt), "GetByID with minimal projection must hydrate ledger_created_at")
 
 	limit := int32(10)
-	ops, err := m.GetAll(ctx, "id", &limit, nil, DESC)
-	require.NoError(t, err)
-	require.Len(t, ops, 1)
-	assert.True(t, now.Equal(ops[0].Operation.LedgerCreatedAt), "GetAll with minimal projection must hydrate ledger_created_at")
 
 	batched, err := m.BatchGetByToID(ctx, 4096, now, "id", &limit, nil, ASC)
 	require.NoError(t, err)

internal/data/query_utils.go

@@ -49,17 +49,6 @@ const (
 	DESC SortOrder = "DESC"
 )
 
-// validatePositiveLimit rejects a non-nil limit that is zero or negative. A caller-supplied
-// page size can go negative (e.g. an int32-overflowed `first` upstream), and silently dropping
-// the LIMIT clause in that case turns a bounded page read into an unbounded full-table read.
-// limit == nil still means "no limit" (used internally) and is left untouched.
-func validatePositiveLimit(limit *int32) error {
-	if limit != nil && *limit <= 0 {
-		return fmt.Errorf("limit must be positive, got %d", *limit)
-	}
-	return nil
-}
-
 // pgtypeTextFromNullString converts sql.NullString to pgtype.Text for efficient binary COPY.
 func pgtypeTextFromNullString(ns sql.NullString) pgtype.Text {
 	return pgtype.Text{String: ns.String, Valid: ns.Valid}

internal/data/sep41/allowances.go

@@ -88,7 +88,7 @@ func (m *AllowanceModel) GetByOwner(ctx context.Context, ownerAddress string, li
 		INNER JOIN contract_tokens ct ON ct.id = a.contract_id
 		WHERE a.owner_id = $1
 		  AND a.expiration_ledger >= COALESCE(
-		      (SELECT value::integer FROM ingest_store WHERE key = $2),
+		      (SELECT value::bigint FROM ingest_store WHERE key = $2),
 		      0
 		  )`
 	args := []interface{}{types.AddressBytea(ownerAddress), latestIngestLedgerCursor}

internal/data/statechanges.go

@@ -124,67 +124,6 @@ func (m *StateChangeModel) BatchGetByAccountAddress(ctx context.Context, account
 	return stateChanges, nil
 }
 
-func (m *StateChangeModel) GetAll(ctx context.Context, columns string, limit *int32, cursor *types.StateChangeCursor, sortOrder SortOrder) ([]*types.StateChangeWithCursor, error) {
-	if err := validatePositiveLimit(limit); err != nil {
-		return nil, err
-	}
-	columns = prepareColumnsWithID(columns, types.StateChange{}, "", "to_id", "operation_id", "state_change_id", "account_id", "ledger_created_at")
-	var queryBuilder strings.Builder
-	var args []interface{}
-	argIndex := 1
-
-	fmt.Fprintf(&queryBuilder, `
-		SELECT %s, ledger_created_at as cursor_ledger_created_at, to_id as cursor_to_id, operation_id as cursor_operation_id, state_change_id as cursor_state_change_id
-		FROM state_changes
-	`, columns)
-
-	// Decomposed cursor pagination: expands ROW() tuple comparison into OR clauses so
-	// TimescaleDB ColumnarScan can push filters into vectorized batch processing.
-	if cursor != nil {
-		clause, cursorArgs, nextIdx := buildDecomposedCursorCondition([]CursorColumn{
-			{Name: "ledger_created_at", Value: cursor.LedgerCreatedAt},
-			{Name: "to_id", Value: cursor.ToID},
-			{Name: "operation_id", Value: cursor.OperationID},
-			{Name: "state_change_id", Value: cursor.StateChangeID},
-		}, sortOrder, argIndex)
-		queryBuilder.WriteString(" WHERE " + clause)
-		args = append(args, cursorArgs...)
-		argIndex = nextIdx
-	}
-
-	// Order with ledger_created_at as leading column for TimescaleDB ChunkAppend
-	if sortOrder == DESC {
-		queryBuilder.WriteString(" ORDER BY ledger_created_at DESC, to_id DESC, operation_id DESC, state_change_id DESC")
-	} else {
-		queryBuilder.WriteString(" ORDER BY ledger_created_at ASC, to_id ASC, operation_id ASC, state_change_id ASC")
-	}
-
-	if limit != nil {
-		fmt.Fprintf(&queryBuilder, " LIMIT $%d", argIndex)
-		args = append(args, *limit)
-	}
-
-	query := queryBuilder.String()
-
-	// For backward pagination, wrap query to reverse the final order.
-	// We use cursor alias columns (e.g., "cursor_ledger_created_at") in ORDER BY to avoid
-	// ambiguity since the inner SELECT includes both original columns and cursor aliases.
-	if sortOrder == DESC {
-		query = fmt.Sprintf(`SELECT * FROM (%s) AS statechanges ORDER BY statechanges.cursor_ledger_created_at ASC, statechanges.cursor_to_id ASC, statechanges.cursor_operation_id ASC, statechanges.cursor_state_change_id ASC`, query)
-	}
-
-	start := time.Now()
-	stateChanges, err := db.QueryManyPtrs[types.StateChangeWithCursor](ctx, m.DB, query, args...)
-	duration := time.Since(start).Seconds()
-	m.Metrics.QueryDuration.WithLabelValues("GetAll", "state_changes").Observe(duration)
-	m.Metrics.QueriesTotal.WithLabelValues("GetAll", "state_changes").Inc()
-	if err != nil {
-		m.Metrics.QueryErrors.WithLabelValues("GetAll", "state_changes", utils.GetDBErrorType(err)).Inc()
-		return nil, fmt.Errorf("getting all state changes: %w", err)
-	}
-	return stateChanges, nil
-}
-
 // BatchCopy inserts state changes using pgx's binary COPY protocol.
 // Uses native pgtype types for optimal performance (see https://github.com/jackc/pgx/issues/763).
 //

internal/data/statechanges_test.go

@@ -610,61 +610,6 @@ func TestStateChangeModel_BatchGetByAccountAddress_DuplicateHashTolerated(t *tes
 	assert.True(t, toIDsFound[2])
 }
 
-func TestStateChangeModel_GetAll(t *testing.T) {
-	dbt := dbtest.Open(t)
-	defer dbt.Close()
-	ctx := context.Background()
-	dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN)
-	require.NoError(t, err)
-	defer dbConnectionPool.Close()
-
-	reg := prometheus.NewRegistry()
-	dbMetrics := metrics.NewMetrics(reg).DB
-
-	m := &StateChangeModel{
-		DB:      dbConnectionPool,
-		Metrics: dbMetrics,
-	}
-
-	now := time.Now()
-
-	address := keypair.MustRandom().Address()
-
-	// Create test transactions first (hash is BYTEA)
-	testHash1 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000001")
-	testHash2 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000002")
-	testHash3 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000003")
-	_, err = dbConnectionPool.Exec(ctx, `
-		INSERT INTO transactions (hash, to_id, fee_charged, result_code, ledger_number, ledger_created_at, is_fee_bump)
-		VALUES
-			($2, 1, 100, 'TransactionResultCodeTxSuccess', 1, $1, false),
-			($3, 2, 200, 'TransactionResultCodeTxSuccess', 2, $1, true),
-			($4, 3, 300, 'TransactionResultCodeTxSuccess', 3, $1, false)
-	`, now, testHash1, testHash2, testHash3)
-	require.NoError(t, err)
-
-	// Create test state changes
-	_, err = dbConnectionPool.Exec(ctx, `
-		INSERT INTO state_changes (to_id, state_change_id, state_change_category, state_change_reason, ledger_created_at, ledger_number, account_id, operation_id)
-		VALUES
-			(1, 1, 'BALANCE', 'CREDIT', $1, 1, $2, 123),
-			(2, 1, 'BALANCE', 'CREDIT', $1, 2, $2, 456),
-			(3, 1, 'BALANCE', 'CREDIT', $1, 3, $2, 789)
-	`, now, types.AddressBytea(address))
-	require.NoError(t, err)
-
-	// Test GetAll without limit
-	stateChanges, err := m.GetAll(ctx, "", nil, nil, DESC)
-	require.NoError(t, err)
-	assert.Len(t, stateChanges, 3)
-
-	// Test GetAll with limit
-	limit := int32(2)
-	stateChanges, err = m.GetAll(ctx, "", &limit, nil, DESC)
-	require.NoError(t, err)
-	assert.Len(t, stateChanges, 2)
-}
-
 func TestStateChangeModel_BatchGetByToIDs(t *testing.T) {
 	dbt := dbtest.Open(t)
 	defer dbt.Close()