diff --git a/harnesses/aggregator-head-lag/scrape-cookie b/harnesses/aggregator-head-lag/scrape-cookie new file mode 100755 index 00000000..585d7baf Binary files /dev/null and b/harnesses/aggregator-head-lag/scrape-cookie differ diff --git a/harnesses/aggregator-head-lag/test-utls b/harnesses/aggregator-head-lag/test-utls new file mode 100755 index 00000000..549c8db9 Binary files /dev/null and b/harnesses/aggregator-head-lag/test-utls differ diff --git a/harnesses/solana-exec/cmd/api/main.go b/harnesses/solana-exec/cmd/api/main.go index 3fcf7cac..491b702e 100644 --- a/harnesses/solana-exec/cmd/api/main.go +++ b/harnesses/solana-exec/cmd/api/main.go @@ -62,48 +62,67 @@ func handleExecLeaderboard(pool *pgxpool.Pool) http.HandlerFunc { defer cancel() rows, err := pool.Query(ctx, ` + WITH cu AS ( + SELECT platform, + percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '24 hours') AS h24_p50, + percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '24 hours') AS h24_p95, + percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '7 days') AS d7_p50, + percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '7 days') AS d7_p95, + percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '30 days') AS d30_p50, + percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro) + FILTER (WHERE block_time >= now() - INTERVAL '30 days') AS d30_p95 + FROM solana_exec_cu_samples + GROUP BY platform + ) SELECT - platform, + f.platform, -- weighted averages: SUM(avg×count)/SUM(count) avoids skewing by small off-peak buckets - SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p95, - SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_pfee, - SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_jito, - SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_count, - - SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p95, - SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_pfee, - SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_jito, - SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_count, - - SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p95, - SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_pfee, - SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_jito, - SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') - / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_count, - - MAX(bucket_start)::text AS latest_bucket - FROM solana_exec_facts - GROUP BY platform`, + SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_prio, + MAX(c.h24_p50) AS h24_p50, + MAX(c.h24_p95) AS h24_p95, + SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_pfee, + SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_jito, + SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_cu, + SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') AS h24_count, + + SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_prio, + MAX(c.d7_p50) AS d7_p50, + MAX(c.d7_p95) AS d7_p95, + SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_pfee, + SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_jito, + SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_cu, + SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') AS d7_count, + + SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_prio, + MAX(c.d30_p50) AS d30_p50, + MAX(c.d30_p95) AS d30_p95, + SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_pfee, + SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_jito, + SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_cu, + SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') AS d30_count, + + MAX(f.bucket_start)::text AS latest_bucket + FROM solana_exec_facts f + LEFT JOIN cu c USING (platform) + WHERE f.bucket_start >= now() - INTERVAL '31 days' + GROUP BY f.platform`, ) if err != nil { log.Printf("exec-api: query: %v", err) diff --git a/harnesses/solana-exec/cmd/materializer/main.go b/harnesses/solana-exec/cmd/materializer/main.go index 98fc753a..63e0c462 100644 --- a/harnesses/solana-exec/cmd/materializer/main.go +++ b/harnesses/solana-exec/cmd/materializer/main.go @@ -27,6 +27,12 @@ func main() { log.Printf("materializer: %s: done", plt) } } + if err := db.PurgeCUSamples(ctx); err != nil { + log.Printf("materializer: purge cu samples: %v", err) + } + if err := db.PurgeEvents(ctx); err != nil { + log.Printf("materializer: purge events: %v", err) + } time.Sleep(5 * time.Minute) } } diff --git a/harnesses/solana-exec/internal/helius/client.go b/harnesses/solana-exec/internal/helius/client.go index 45bc6c17..29433118 100644 --- a/harnesses/solana-exec/internal/helius/client.go +++ b/harnesses/solana-exec/internal/helius/client.go @@ -10,22 +10,32 @@ import ( ) // Client wraps the Helius RPC + enhanced-transactions APIs. +// rpcURL is used for standard Solana JSON-RPC calls (getSignaturesForAddress) and can be +// any Solana endpoint — we default to the public mainnet RPC to avoid consuming Helius credits +// on raw pagination. enhURL is Helius-specific and costs credits; it is only called for the +// 100-sig quality sample per poll. type Client struct { httpClient *http.Client apiKey string - rpcURL string // https://mainnet.helius-rpc.com/?api-key=KEY - enhURL string // https://api.helius.xyz/v0/transactions?api-key=KEY + rpcURL string // standard Solana JSON-RPC (public endpoint, no credits) + enhURL string // https://api.helius.xyz/v0/transactions?api-key=KEY (credits) } -func New(apiKey string) *Client { +// NewWithRPC creates a client where rpcURL is used for standard RPC calls (getSignaturesForAddress). +// Pass a public or self-hosted endpoint to avoid consuming Helius credits on pagination. +func NewWithRPC(apiKey, rpcURL string) *Client { return &Client{ httpClient: &http.Client{Timeout: 30 * time.Second}, apiKey: apiKey, - rpcURL: fmt.Sprintf("https://mainnet.helius-rpc.com/?api-key=%s", apiKey), + rpcURL: rpcURL, enhURL: fmt.Sprintf("https://api.helius.xyz/v0/transactions?api-key=%s", apiKey), } } +func New(apiKey string) *Client { + return NewWithRPC(apiKey, "https://api.mainnet-beta.solana.com") +} + // SigEntry is one result from getSignaturesForAddress. type SigEntry struct { Signature string `json:"signature"` @@ -57,36 +67,57 @@ func (c *Client) GetSignaturesForAddress(ctx context.Context, address string, li "params": []any{address, params}, }) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.rpcURL, bytes.NewReader(body)) - if err != nil { - return nil, err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("helius RPC HTTP %d", resp.StatusCode) - } - - var out struct { - Result []SigEntry `json:"result"` - Error *struct { - Code int `json:"code"` - Message string `json:"message"` - } `json:"error"` - } - if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { - return nil, fmt.Errorf("helius RPC decode: %w", err) - } - if out.Error != nil { - return nil, fmt.Errorf("helius RPC error %d: %s", out.Error.Code, out.Error.Message) - } - return out.Result, nil + // Retry up to 3 times on 429 or transient errors with exponential backoff. + var lastErr error + for attempt := range 3 { + if attempt > 0 { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(time.Duration(attempt*attempt) * time.Second): + } + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.rpcURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + lastErr = err + continue + } + + if resp.StatusCode == http.StatusTooManyRequests { + resp.Body.Close() + lastErr = fmt.Errorf("helius RPC HTTP 429") + continue + } + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, fmt.Errorf("helius RPC HTTP %d", resp.StatusCode) + } + + var out struct { + Result []SigEntry `json:"result"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + resp.Body.Close() + return nil, fmt.Errorf("helius RPC decode: %w", err) + } + resp.Body.Close() + if out.Error != nil { + return nil, fmt.Errorf("helius RPC error %d: %s", out.Error.Code, out.Error.Message) + } + return out.Result, nil + } + return nil, fmt.Errorf("helius RPC: %w (after 3 attempts)", lastErr) } // NativeTransfer is a SOL transfer extracted by Helius. diff --git a/harnesses/solana-exec/migrations/003_cu_samples.sql b/harnesses/solana-exec/migrations/003_cu_samples.sql new file mode 100644 index 00000000..38d89388 --- /dev/null +++ b/harnesses/solana-exec/migrations/003_cu_samples.sql @@ -0,0 +1,16 @@ +BEGIN; + +-- Raw CU price samples from the 100-sig enhanced API poll each cycle. +-- Gives real percentile_cont on actual data points instead of AVG(hourly_p50). +-- Retention: 35 days (purged by materializer). +CREATE TABLE IF NOT EXISTS solana_exec_cu_samples ( + id BIGSERIAL PRIMARY KEY, + platform TEXT NOT NULL, + block_time TIMESTAMPTZ NOT NULL, + cu_price_micro BIGINT NOT NULL +); + +CREATE INDEX IF NOT EXISTS solana_exec_cu_samples_platform_time + ON solana_exec_cu_samples (platform, block_time); + +COMMIT; diff --git a/harnesses/solana-exec/migrations/004_raw_counts_idempotent.sql b/harnesses/solana-exec/migrations/004_raw_counts_idempotent.sql new file mode 100644 index 00000000..66ab942a --- /dev/null +++ b/harnesses/solana-exec/migrations/004_raw_counts_idempotent.sql @@ -0,0 +1,17 @@ +BEGIN; + +-- Make raw_counts idempotent on crash/restart. +-- Problem: UpsertRawCounts accumulated (success_count + EXCLUDED.success_count). +-- On a crash between UpsertRawCounts and SaveCursor, the next boot re-processes +-- the same sigs and double-counts. Fix: key by (platform, bucket_start, from_cursor) +-- so each poll's contribution is a distinct row. ON CONFLICT DO NOTHING = fully idempotent. + +ALTER TABLE solana_exec_raw_counts DROP CONSTRAINT solana_exec_raw_counts_pkey; + +-- from_cursor = cursor.LastSig at the start of the poll that produced this row. +-- Existing rows get '' which is safe — they predate this schema change and won't be re-inserted. +ALTER TABLE solana_exec_raw_counts ADD COLUMN IF NOT EXISTS from_cursor TEXT NOT NULL DEFAULT ''; + +ALTER TABLE solana_exec_raw_counts ADD PRIMARY KEY (platform, bucket_start, from_cursor); + +COMMIT; diff --git a/harnesses/solana-exec/migrations/005_events_purge_and_cu_sig.sql b/harnesses/solana-exec/migrations/005_events_purge_and_cu_sig.sql new file mode 100644 index 00000000..2594ba92 --- /dev/null +++ b/harnesses/solana-exec/migrations/005_events_purge_and_cu_sig.sql @@ -0,0 +1,18 @@ +BEGIN; + +-- Bound events table growth: materializer will purge rows older than 90 days. +-- Facts table already holds all aggregated metrics; raw events are only needed +-- for re-materialization within the rolling window. +-- No schema change needed — purge is done in application code. + +-- Make cu_samples idempotent on crash/restart by keying on (platform, sig). +-- Without this, a crash between UpsertEvents and SaveCursor causes the next +-- boot to re-call InsertCUSamples for the same 100 sigs, duplicating them +-- and slightly skewing percentiles. +ALTER TABLE solana_exec_cu_samples ADD COLUMN IF NOT EXISTS sig TEXT; + +CREATE UNIQUE INDEX IF NOT EXISTS solana_exec_cu_samples_platform_sig + ON solana_exec_cu_samples (platform, sig) + WHERE sig IS NOT NULL; + +COMMIT; diff --git a/src/app/api/internal/codex-token/route.ts b/src/app/api/internal/codex-token/route.ts new file mode 100644 index 00000000..c921b979 --- /dev/null +++ b/src/app/api/internal/codex-token/route.ts @@ -0,0 +1,68 @@ +export const runtime = 'edge'; + +const SECRET = process.env.INTERNAL_RELAY_SECRET ?? ''; + +export async function GET(request: Request) { + const auth = request.headers.get('x-relay-secret'); + if (SECRET && auth !== SECRET) { + return new Response('Forbidden', { status: 403 }); + } + + const body = JSON.stringify({ + operationName: 'CreateApiToken', + query: 'mutation CreateApiToken { createApiTokens(input: { count: 1 }) { token } }', + variables: {}, + }); + + const resp = await fetch('https://un.defined.fi/api', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json', + Origin: 'https://un.defined.fi', + Referer: 'https://un.defined.fi/', + 'User-Agent': + 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36', + 'Accept-Language': 'en-US,en;q=0.9', + 'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="131", "Google Chrome";v="131"', + 'sec-ch-ua-mobile': '?0', + 'sec-ch-ua-platform': '"macOS"', + 'sec-fetch-dest': 'empty', + 'sec-fetch-mode': 'cors', + 'sec-fetch-site': 'same-origin', + }, + body, + }); + + const status = resp.status; + const text = await resp.text(); + + if (status !== 200) { + return new Response(JSON.stringify({ error: `upstream ${status}`, body: text.slice(0, 200) }), { + status: 502, + headers: { 'Content-Type': 'application/json' }, + }); + } + + let parsed: { data?: { createApiTokens?: { token: string }[] } }; + try { + parsed = JSON.parse(text); + } catch { + return new Response(JSON.stringify({ error: 'parse error', body: text.slice(0, 200) }), { + status: 502, + headers: { 'Content-Type': 'application/json' }, + }); + } + + const token = parsed?.data?.createApiTokens?.[0]?.token; + if (!token) { + return new Response(JSON.stringify({ error: 'no token', body: text.slice(0, 300) }), { + status: 502, + headers: { 'Content-Type': 'application/json' }, + }); + } + + return new Response(token, { + headers: { 'Content-Type': 'text/plain' }, + }); +}