feat(solana-exec): multi-platform support — pump.fun, photon, bullx, GMGN, Axiom, FOMO

harnesses/solana-exec/cmd/collector/main.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sort"
+	"strings"
 	"time"
 
 	"github.com/ChainBench/OpenChainBench/harnesses/solana-exec/internal/helius"
@@ -24,90 +26,147 @@ func main() {
 	}
 	defer db.Close()
 
-	h := helius.New(mustEnv("HELIUS_API_KEY"))
+	rpcURL := os.Getenv("SOLANA_RPC_URL")
+	if rpcURL == "" {
+		rpcURL = "https://api.mainnet-beta.solana.com"
+	}
+	h := helius.New(rpcURL)
 
 	log.Printf("collector: monitoring %d platforms", len(platform.FeeAccounts))
 
 	for {
-		for plt, feeAccount := range platform.FeeAccounts {
-			if err := collect(ctx, db, h, plt, feeAccount); err != nil {
+		for plt, feeAccounts := range platform.FeeAccounts {
+			if err := collect(ctx, db, h, plt, feeAccounts); err != nil {
 				log.Printf("collector: %s: %v", plt, err)
 			}
 		}
 		time.Sleep(10 * time.Minute)
 	}
 }
 
-func collect(ctx context.Context, db *store.DB, h *helius.Client, plt, feeAccount string) error {
-	cursor, err := db.GetCursor(ctx, plt)
-	if err != nil {
-		return fmt.Errorf("get cursor: %w", err)
+type acctData struct {
+	cursorKey string
+	cursor    store.Cursor
+	sigs      []helius.SigEntry // newest-first from RPC
+}
+
+func collect(ctx context.Context, db *store.DB, h *helius.Client, plt string, feeAccounts []string) error {
+	// SPL token mint for fee detection; empty = SOL (native lamports).
+	feeToken := platform.FeeToken[plt]
+
+	// feeSet is used to detect fee transfers in NativeTransfers / postTokenBalances.owner.
+	// For SPL-token platforms (e.g. FOMO), FeeAccounts holds the ATA (for sig discovery)
+	// but postTokenBalances.owner is the wallet — so use FeeOwners when present.
+	feeOwners := platform.FeeOwners[plt]
+	if feeOwners == nil {
+		feeOwners = feeAccounts
+	}
+	feeSet := make(map[string]bool, len(feeOwners))
+	for _, fa := range feeOwners {
+		feeSet[fa] = true
 	}
 
-	const sigLimit = 100
-	// Paginate through ALL signatures newer than cursor (newest-first per page).
-	// Each page uses `before=oldestSigInPreviousPage` to walk backwards until
-	// we exhaust the window. This guarantees complete coverage regardless of volume.
-	var sigs []helius.SigEntry
-	before := ""
-	for {
-		batch, err := h.GetSignaturesForAddress(ctx, feeAccount, sigLimit, cursor.LastSig, before)
+	// Paginate each fee account independently with its own cursor so a crash
+	// between accounts doesn't cause re-processing of already-committed data.
+	perAcct := make([]acctData, 0, len(feeAccounts))
+
+	for _, feeAccount := range feeAccounts {
+		cursorKey := plt + ":" + feeAccount
+		cursor, err := db.GetCursor(ctx, cursorKey)
 		if err != nil {
-			return fmt.Errorf("get sigs (before=%s): %w", before, err)
+			return fmt.Errorf("get cursor %s: %w", feeAccount, err)
 		}
-		sigs = append(sigs, batch...)
-		if len(batch) < sigLimit {
-			break // last page
+
+		const sigLimit = 1000
+		var sigs []helius.SigEntry
+		before := ""
+		for {
+			batch, err := h.GetSignaturesForAddress(ctx, feeAccount, sigLimit, cursor.LastSig, before)
+			if err != nil {
+				return fmt.Errorf("get sigs %s (before=%s): %w", feeAccount, before, err)
+			}
+			sigs = append(sigs, batch...)
+			if len(batch) < sigLimit {
+				break
+			}
+			before = batch[len(batch)-1].Signature
+			time.Sleep(100 * time.Millisecond)
+		}
+		if len(sigs) > sigLimit {
+			log.Printf("collector: %s: %s: paginated %d sigs", plt, feeAccount, len(sigs))
 		}
-		before = batch[len(batch)-1].Signature
-		time.Sleep(300 * time.Millisecond) // respect Helius free-tier rate limit between pages
-	}
-	if len(sigs) == 0 {
-		return nil
-	}
-	if len(sigs) > sigLimit {
-		log.Printf("collector: %s: paginated %d raw sigs (%d pages)", plt, len(sigs), (len(sigs)+sigLimit-1)/sigLimit)
-	}
 
-	// Reverse to process oldest-first so cursor is always the true watermark.
-	reversed := make([]helius.SigEntry, len(sigs))
-	for i, s := range sigs {
-		reversed[len(sigs)-1-i] = s
+		perAcct = append(perAcct, acctData{cursorKey: cursorKey, cursor: cursor, sigs: sigs})
 	}
 
-	// Pre-filter failed txs to avoid wasting enhanced-API credits.
-	// Failed txs don't generate platform fees; priority-fee stats should reflect
-	// the cost of successful trades only.
-	var sigStrs []string
-	for _, s := range reversed {
-		if s.Err == nil {
-			sigStrs = append(sigStrs, s.Signature)
+	// Merge sigs from all accounts, de-duplicate by signature.
+	// A single tx can touch multiple fee accounts of the same platform (rare but possible).
+	seen := make(map[string]struct{})
+	var allSigs []helius.SigEntry
+	for _, acct := range perAcct {
+		for _, s := range acct.sigs {
+			if _, dup := seen[s.Signature]; dup {
+				continue
+			}
+			seen[s.Signature] = struct{}{}
+			allSigs = append(allSigs, s)
 		}
 	}
 
-	// Count successful sigs per hour bucket from raw pagination (full volume, no enhanced API cost).
+	if len(allSigs) == 0 {
+		return nil
+	}
+
+	// Count raw hourly volume from the merged+deduped sig set so a tx touching
+	// multiple fee accounts of the same platform is counted exactly once.
+	// from_cursor is a compound of all account cursors for idempotency on retry.
+	fromCursor := compoundCursor(perAcct)
 	hourBuckets := make(map[time.Time]int64)
-	for _, s := range reversed {
+	for _, s := range allSigs {
 		if s.Err != nil || s.BlockTime == 0 {
 			continue
 		}
 		bucket := time.Unix(s.BlockTime, 0).UTC().Truncate(time.Hour)
 		hourBuckets[bucket]++
 	}
-	if err := db.UpsertRawCounts(ctx, plt, hourBuckets); err != nil {
+	if err := db.UpsertRawCounts(ctx, plt, fromCursor, hourBuckets); err != nil {
 		return fmt.Errorf("upsert raw counts: %w", err)
 	}
 
+	// Pre-filter failed txs; only successful trades generate platform fees.
+	var sigStrs []string
+	for _, s := range allSigs {
+		if s.Err == nil {
+			sigStrs = append(sigStrs, s.Signature)
+		}
+	}
+
+	// All sigs failed: still save cursors so we don't re-scan next poll.
 	if len(sigStrs) == 0 {
-		// All sigs in this batch were failed txs; advance cursor and skip.
-		newest := sigs[0]
-		return db.SaveCursor(ctx, plt, newest.Signature, newest.Slot)
+		return saveCursors(ctx, db, perAcct)
+	}
+
+	// Sort by slot descending before striding so the sample is temporally uniform
+	// across multi-account platforms (e.g. Axiom 20 accounts — merge order is arbitrary).
+	sort.Slice(allSigs, func(i, j int) bool { return allSigs[i].Slot > allSigs[j].Slot })
+
+	// Rebuild sigStrs from the sorted order after filtering failed txs.
+	sigStrs = sigStrs[:0]
+	for _, s := range allSigs {
+		if s.Err == nil {
+			sigStrs = append(sigStrs, s.Signature)
+		}
 	}
 
-	// Cap enhanced API at 100 sigs per poll (Helius free-tier budget: ~432K CUs/month).
-	// Fee quality metrics are sampled; tx_count comes from raw counts above.
+	// Cap enhanced API at 100 sigs; stride-sample across the full window so the
+	// sample represents the whole poll period, not just the most recent transactions.
 	if len(sigStrs) > 100 {
-		sigStrs = sigStrs[:100]
+		step := len(sigStrs) / 100
+		sampled := make([]string, 0, 100)
+		for i := 0; i < len(sigStrs) && len(sampled) < 100; i += step {
+			sampled = append(sampled, sigStrs[i])
+		}
+		sigStrs = sampled
 	}
 
 	txs, err := h.GetEnhancedTransactions(ctx, sigStrs)
@@ -128,20 +187,39 @@ func collect(ctx context.Context, db *store.DB, h *helius.Client, plt, feeAccoun
 
 		var platformFeeLamports, jitoTipLamports int64
 		isJito := false
-		for _, xfer := range tx.NativeTransfers {
-			if xfer.ToUserAccount == feeAccount {
-				platformFeeLamports += xfer.Amount
+
+		if feeToken == "" {
+			// SOL-fee platform: detect via native SOL transfers.
+			for _, xfer := range tx.NativeTransfers {
+				if feeSet[xfer.ToUserAccount] {
+					platformFeeLamports += xfer.Amount
+				}
+				if platform.JitoTipAccounts[xfer.ToUserAccount] {
+					jitoTipLamports += xfer.Amount
+					isJito = true
+				}
+			}
+		} else {
+			// SPL-token-fee platform (e.g. FOMO uses USDC): detect via token transfers.
+			// platformFeeLamports stores raw token units (1 USDC = 1_000_000).
+			for _, xfer := range tx.TokenTransfers {
+				if feeSet[xfer.ToOwner] && xfer.Mint == feeToken {
+					platformFeeLamports += xfer.Amount
+				}
 			}
-			if platform.JitoTipAccounts[xfer.ToUserAccount] {
-				jitoTipLamports += xfer.Amount
-				isJito = true
+			// Jito detection is SOL-based regardless of fee token.
+			for _, xfer := range tx.NativeTransfers {
+				if platform.JitoTipAccounts[xfer.ToUserAccount] {
+					jitoTipLamports += xfer.Amount
+					isJito = true
+				}
 			}
 		}
 
-		var cuPriceMicro int64
-		if tx.ComputeUnitsConsumed > 0 && priorityFee > 0 {
-			cuPriceMicro = priorityFee * 1_000_000 / tx.ComputeUnitsConsumed
-		}
+		// CU price comes directly from SetComputeUnitPrice (disc 0x03), already in
+		// micro-lamports per CU — no division needed. Only set when the instruction
+		// is present; 0 means base fee only, not a zero-price priority tx.
+		cuPriceMicro := tx.CUPriceDeclared
 
 		events = append(events, store.ExecEvent{
 			Sig:                 tx.Signature,
@@ -162,13 +240,48 @@ func collect(ctx context.Context, db *store.DB, h *helius.Client, plt, feeAccoun
 		return fmt.Errorf("upsert: %w", err)
 	}
 
-	// Advance cursor to the newest sig (first in original order = last in reversed).
-	newest := sigs[0]
-	if err := db.SaveCursor(ctx, plt, newest.Signature, newest.Slot); err != nil {
-		return fmt.Errorf("save cursor: %w", err)
+	cuSamples := make([]store.CUSample, 0, len(events))
+	for _, e := range events {
+		if e.CUPriceMicro > 0 {
+			cuSamples = append(cuSamples, store.CUSample{Sig: e.Sig, BlockTime: e.BlockTime, CUPriceMicro: e.CUPriceMicro})
+		}
+	}
+	if err := db.InsertCUSamples(ctx, plt, cuSamples); err != nil {
+		return fmt.Errorf("insert cu samples: %w", err)
 	}
 
-	log.Printf("collector: %s: ingested %d txs (newest slot %d)", plt, len(events), newest.Slot)
+	if err := saveCursors(ctx, db, perAcct); err != nil {
+		return err
+	}
+
+	log.Printf("collector: %s: ingested %d events", plt, len(events))
+	return nil
+}
+
+// compoundCursor returns a deterministic string that encodes the cursor state for
+// all accounts in this poll, used as the from_cursor idempotency key in raw_counts.
+func compoundCursor(perAcct []acctData) string {
+	if len(perAcct) == 0 {
+		return ""
+	}
+	parts := make([]string, 0, len(perAcct))
+	for _, a := range perAcct {
+		parts = append(parts, a.cursorKey+"="+a.cursor.LastSig)
+	}
+	return strings.Join(parts, ",")
+}
+
+// saveCursors advances each fee account's cursor to its newest observed sig.
+func saveCursors(ctx context.Context, db *store.DB, perAcct []acctData) error {
+	for _, acct := range perAcct {
+		if len(acct.sigs) == 0 {
+			continue
+		}
+		newest := acct.sigs[0] // RPC returns newest-first
+		if err := db.SaveCursor(ctx, acct.cursorKey, newest.Signature, newest.Slot); err != nil {
+			return fmt.Errorf("save cursor %s: %w", acct.cursorKey, err)
+		}
+	}
 	return nil
 }
 

harnesses/solana-exec/docker-compose.solana-exec.yml

@@ -1,5 +1,6 @@
 # Fragment to merge into /opt/ocb/docker-compose.yml on VPS.
-# Requires: ocb-postgres (from docker-compose.apps.yml), HELIUS_API_KEY in env.
+# Requires: ocb-postgres (from docker-compose.apps.yml).
+# Optional: SOLANA_RPC_URL (defaults to api.mainnet-beta.solana.com).
 
 services:
   solana-exec-collector: