Chore/history UI improvements

nt-be/.env.example

@@ -32,6 +32,8 @@ PORT=3000
 # Feature flags
 # DISABLE_STAKING_REWARDS=true  # Disables epoch-snapshot + gap-fill binary search (saves RPC)
 # DISABLE_FT_LOCKUP_SCHEDULER=true  # Disables FT lockup schedule refresh + claim worker
+# PUBLIC_HISTORY_MEDALLION_READS=false  # true routes backfill-complete public DAOs to gold
+# GOLD_USD_ENRICHMENT_INTERVAL_SECONDS=300  # ordered historical-price load then NULL USD fill
 
 # Telegram Bot (optional — notifications degrade gracefully without it)
 # TELEGRAM_BOT_TOKEN=your_bot_token_here

nt-be/Cargo.toml

@@ -8,6 +8,7 @@ default-run = "nt-be"
 async-trait = "0.1"
 axum = { version = "0.8.7", features = ["macros"] }
 tokio = { version = "1", features = ["full"] }
+tokio-util = "0.7"
 tower-http = { version = "0.6", features = ["cors", "trace"] }
 near-api = "0.8"
 near-account-id = "2.0"

nt-be/docs/BACKGROUND_JOBS.md

@@ -17,10 +17,23 @@ What this replaces: 17 hand-rolled `tokio::spawn` + interval loops in
 
 ## Resilience
 
-All workers run under a single apalis `Monitor` (`spawn_all` in
-`src/jobs/mod.rs`) — the framework's own multi-worker supervisor, rather
-than a hand-rolled `tokio::spawn` loop per worker. It gives three layers of
-failure containment plus coordinated shutdown:
+Every web replica prepares the Apalis schema and queue registry, but only the
+replica holding the PostgreSQL session advisory lock `(0x54525A55, 1)` starts
+workers. The elected replica holds one pooled connection for its complete
+leadership session; followers release unsuccessful connections immediately,
+continue serving HTTP, and retry election with jittered exponential backoff
+(2s→30s). PostgreSQL automatically releases the lock if the leader process or
+connection dies.
+
+The leader runs all cron workers under one Apalis `Monitor` and starts the two
+public-history payload consumers (`public-history-latest` and
+`public-history-backfill`) in the same leadership session. This prevents a
+rolling deployment from creating a second scheduler or a second fixed-name
+payload worker. Startup tasks are pushed only on the first leadership
+acquisition per process and are skipped when an active task for that queue
+already exists.
+
+Failure containment and coordinated shutdown have four layers:
 
 1. **Task errors** — the failure is recorded and the next cron tick starts
    a fresh task. One bad cycle never stops the schedule. There is
@@ -38,14 +51,43 @@ failure containment plus coordinated shutdown:
    a restart. The whole set runs on one supervised task and stops together
    on **`SIGINT` or `SIGTERM`** (`run_with_signal`) — SIGTERM being what
    containers/systemd send for a graceful stop.
-
-Restarts are immediate (apalis's `Monitor`'s `should_restart` hook is
-synchronous, so it can't back off). This doesn't hot-loop: transient DB
-outages are absorbed by apalis-postgres's own fetcher backoff (1s→5min)
-*inside* `worker.run()`, so a worker rarely exits on a blip; it only exits
-on a terminal condition, which is rare. If a cooldown on repeated exits is
-later wanted, apalis's native `circuit_breaker` worker ext (with a
-`recovery_timeout`) is the tool.
+4. **Replica exit or leadership loss** — a five-second heartbeat verifies the
+   lock-owning connection. A heartbeat failure or unexpected runtime exit
+   cancels and drains the complete worker runtime before leadership is
+   reacquired. Graceful shutdown explicitly unlocks before returning the
+   connection to the pool; any unlock failure closes the connection instead,
+   so a lock-bearing session is never reused.
+
+Cron worker restarts remain immediate because apalis's `Monitor` hook is
+synchronous. The public-history payload consumers instead retry every
+unexpected worker-loop exit with jittered exponential backoff (2s→60s),
+including `WORKER_ALREADY_EXISTS` as a final safeguard and terminal
+database/TLS disconnects. Ordinary task-handler failures remain task results
+inside Apalis and do not restart a consumer. A shared shutdown token cancels
+election/backoff sleeps and lets active workers drain without delaying HTTP
+readiness or shared HTTP clients.
+
+The durable `background_job_leader` row records the current generation and
+heartbeat for operations; `/api/health` reports both the local role and global
+leadership record without making followers unhealthy. The leader reserves one
+slot from the 20-connection application pool, leaving up to 19 for application
+and worker queries.
+
+Leadership guarantees one active scheduler/runtime during normal operation.
+Task processing itself remains at-least-once: after a crash, Apalis may retry a
+task whose completion was not persisted, so handlers still need their existing
+idempotency and deduplication protections.
+
+Public silver and gold projection remain idle for a DAO until all three rows in
+`bronze_public_history_cursors` (`nearblocks_ft`, `nearblocks_mt`, and
+`nearblocks_receipt`) have `backfill_done = true`. Public API reads stay on
+legacy `balance_changes` by default; setting `PUBLIC_HISTORY_MEDALLION_READS=true`
+routes backfill-complete public DAOs to `gold_public_history_events`.
+
+Gold projection values recent ordinary transfers from the in-memory latest-price
+snapshot, without a database or external request. Historical rows and recent
+rows whose token has no cached price keep NULL USD amounts until the ordered
+`gold-usd-enrichment` job fills them. Exchange quote USD remains authoritative.
 
 Handlers that do several steps run **all** steps and aggregate failures
 rather than aborting on the first error (e.g. notifications detect +
@@ -94,9 +136,10 @@ returning a plain `404` (the board only owns `/api/v1`).
 | confidential-poll | every 300s | CONFIDENTIAL_POLL_INTERVAL_SECONDS | gated by DISABLE_BALANCE_MONITORING |
 | price-sync | every 60s | — | syncs DeFiLlama prices into `historical_prices` |
 | token-price-ingest | every 60s | — | refreshes `tokens` and 5-minute `token_prices` from Chaindefuser |
+| gold-usd-enrichment | every 5min + startup task | GOLD_USD_ENRICHMENT_INTERVAL_SECONDS | ordered historical-price load, then NULL public/confidential USD fill; projection never waits for prices |
 | public-history-scheduler | every 2s | — | enqueues public latest/backfill page jobs; latest enqueue skipped without GOLDSKY_DATABASE_URL |
-| public-silver-projection | every 5s | — | gated by NEARBLOCKS_API_KEY |
-| public-gold-projection | every 5s | — | gated by NEARBLOCKS_API_KEY |
+| public-silver-projection | every 5s | — | gated by NEARBLOCKS_API_KEY and per-DAO bronze backfill completion |
+| public-gold-projection | every 5s | — | gated by NEARBLOCKS_API_KEY and per-DAO bronze backfill completion |
 | public-proposal-reconciliation | every 10min | — | gated by NEARBLOCKS_API_KEY |
 | confidential-history-ingest | every 10s | — | |
 | confidential-snapshots | hourly | — | |

nt-be/migrations/20260720000001_create_background_job_leader.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS background_job_leader (
+    lock_name TEXT PRIMARY KEY,
+    instance_id UUID NOT NULL,
+    generation BIGINT NOT NULL,
+    acquired_at TIMESTAMPTZ NOT NULL,
+    heartbeat_at TIMESTAMPTZ NOT NULL,
+    released_at TIMESTAMPTZ
+);

nt-be/migrations/20260720000002_add_public_gold_projection_readiness.sql

@@ -0,0 +1,154 @@
+-- Public medallion reads may start only after a successful gold projection.
+-- Bronze backfill completion alone is not sufficient because silver and gold
+-- are projected asynchronously after the final bronze cursor is committed.
+ALTER TABLE gold_public_history_cursors
+    ADD COLUMN IF NOT EXISTS projection_ready_at TIMESTAMPTZ,
+    ADD COLUMN IF NOT EXISTS projection_ready_version INTEGER NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS projection_validation_pending BOOLEAN NOT NULL DEFAULT false,
+    ADD COLUMN IF NOT EXISTS gold_force_full_recompute BOOLEAN NOT NULL DEFAULT false;
+
+COMMENT ON COLUMN gold_public_history_cursors.projection_ready_at IS
+    'Set by a successful gold projection after bronze backfill and silver projection are complete; cleared whenever derived public history becomes dirty.';
+COMMENT ON COLUMN gold_public_history_cursors.projection_ready_version IS
+    'Readiness schema epoch published by the projector; readers require the current application epoch.';
+COMMENT ON COLUMN gold_public_history_cursors.projection_validation_pending IS
+    'Durable handoff flag consumed by current projectors, including after an older binary clears the legacy dirty timestamp.';
+COMMENT ON COLUMN gold_public_history_cursors.gold_force_full_recompute IS
+    'Preserves a required full rebuild across later incremental dirty marks until a current projector succeeds.';
+
+-- Database-level invalidation protects rolling deploys. Older binaries do not
+-- know about the readiness columns, but their ordinary bronze/silver/gold
+-- cursor writes still clear the marker and leave a durable validation request
+-- for a current binary.
+CREATE OR REPLACE FUNCTION invalidate_public_gold_readiness_on_gold_dirty()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    IF NEW.gold_dirty_since IS NOT NULL THEN
+        NEW.projection_ready_at := NULL;
+        NEW.projection_ready_version := 0;
+        NEW.projection_validation_pending := true;
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_invalidate_public_gold_readiness_on_gold_dirty
+    ON gold_public_history_cursors;
+CREATE TRIGGER trg_invalidate_public_gold_readiness_on_gold_dirty
+BEFORE INSERT OR UPDATE ON gold_public_history_cursors
+FOR EACH ROW
+EXECUTE FUNCTION invalidate_public_gold_readiness_on_gold_dirty();
+
+CREATE OR REPLACE FUNCTION invalidate_public_gold_readiness_on_silver_dirty()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    IF NEW.silver_dirty_since IS NOT NULL THEN
+        INSERT INTO gold_public_history_cursors (
+            account_id,
+            projection_ready_at,
+            projection_ready_version,
+            projection_validation_pending,
+            updated_at
+        )
+        VALUES (NEW.account_id, NULL, 0, true, NOW())
+        ON CONFLICT (account_id) DO UPDATE SET
+            projection_ready_at = NULL,
+            projection_ready_version = 0,
+            projection_validation_pending = true,
+            updated_at = NOW();
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_invalidate_public_gold_readiness_on_silver_dirty
+    ON silver_public_history_cursors;
+CREATE TRIGGER trg_invalidate_public_gold_readiness_on_silver_dirty
+AFTER INSERT OR UPDATE ON silver_public_history_cursors
+FOR EACH ROW
+EXECUTE FUNCTION invalidate_public_gold_readiness_on_silver_dirty();
+
+CREATE OR REPLACE FUNCTION invalidate_public_gold_readiness_on_backfill_progress()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    INSERT INTO gold_public_history_cursors (
+        account_id,
+        projection_ready_at,
+        projection_ready_version,
+        projection_validation_pending,
+        gold_force_full_recompute,
+        updated_at
+    )
+    VALUES (NEW.account_id, NULL, 0, true, true, NOW())
+    ON CONFLICT (account_id) DO UPDATE SET
+        projection_ready_at = NULL,
+        projection_ready_version = 0,
+        projection_validation_pending = true,
+        gold_force_full_recompute = true,
+        updated_at = NOW();
+    RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_invalidate_public_gold_readiness_on_backfill_insert
+    ON bronze_public_history_cursors;
+CREATE TRIGGER trg_invalidate_public_gold_readiness_on_backfill_insert
+AFTER INSERT ON bronze_public_history_cursors
+FOR EACH ROW
+EXECUTE FUNCTION invalidate_public_gold_readiness_on_backfill_progress();
+
+DROP TRIGGER IF EXISTS trg_invalidate_public_gold_readiness_on_backfill_update
+    ON bronze_public_history_cursors;
+CREATE TRIGGER trg_invalidate_public_gold_readiness_on_backfill_update
+AFTER UPDATE OF backward_cursor, backfill_done ON bronze_public_history_cursors
+FOR EACH ROW
+EXECUTE FUNCTION invalidate_public_gold_readiness_on_backfill_progress();
+
+-- Require one verified post-migration projection before enabling medallion
+-- reads for accounts whose bronze backfill was already complete. A NULL
+-- recompute cursor deliberately requests a full gold rebuild.
+INSERT INTO gold_public_history_cursors (
+    account_id,
+    gold_dirty_since,
+    gold_recompute_from,
+    projection_ready_at,
+    projection_ready_version,
+    projection_validation_pending,
+    gold_force_full_recompute,
+    updated_at
+)
+SELECT
+    completed.account_id,
+    NOW(),
+    NULL,
+    NULL,
+    0,
+    true,
+    true,
+    NOW()
+FROM (
+    SELECT account_id
+    FROM bronze_public_history_cursors
+    WHERE source IN (
+        'nearblocks_ft'::public_history_source,
+        'nearblocks_mt'::public_history_source,
+        'nearblocks_receipt'::public_history_source
+    )
+      AND backfill_done = true
+    GROUP BY account_id
+    HAVING COUNT(*) = 3
+) completed
+ON CONFLICT (account_id) DO UPDATE SET
+    gold_dirty_since = NOW(),
+    gold_recompute_from = NULL,
+    projection_ready_at = NULL,
+    projection_ready_version = 0,
+    projection_validation_pending = true,
+    gold_force_full_recompute = true,
+    updated_at = NOW();

nt-be/src/app_state.rs

@@ -11,6 +11,7 @@ use crate::{
         TransferHintService, fastnear::FastNearProvider, neardata::NeardataClient,
     },
     handlers::public_history::bronze::NearblocksPriority,
+    jobs::leadership::BackgroundJobsStatus,
     services::{DeFiLlamaClient, PriceLookupService, TokenPriceService},
     utils::{
         cache::{Cache, CacheKey, CacheTier},
@@ -106,6 +107,7 @@ pub struct AppState {
     /// None if GOLDSKY_DATABASE_URL is not configured.
     pub goldsky_pool: Option<PgPool>,
     pub event_tx: broadcast::Sender<AppEvent>,
+    pub background_jobs_status: Arc<BackgroundJobsStatus>,
     /// Wakes the treasury creation sweeper immediately (instead of waiting for
     /// its next poll tick) when an attempt fails, so a half-created treasury is
     /// retried within moments rather than seconds.
@@ -431,6 +433,7 @@ impl AppStateBuilder {
             neardata_client,
             goldsky_pool,
             event_tx,
+            background_jobs_status: Arc::new(BackgroundJobsStatus::new()),
             creation_sweep_notify: Arc::new(tokio::sync::Notify::new()),
         })
     }