diff --git a/api/dbv1/models.go b/api/dbv1/models.go index 9c6da6e6..05909b48 100644 --- a/api/dbv1/models.go +++ b/api/dbv1/models.go @@ -1620,16 +1620,6 @@ type MvDashboardTransactionType struct { TransactionCount int64 `json:"transaction_count"` } -// Queue of ManageEntity transactions to be forwarded to the new Core chain (audius-mainnet-v2) during genesis migration. -type NewChainQueue struct { - ID int64 `json:"id"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - // Protobuf-serialized ManageEntityLegacy message. - TxData []byte `json:"tx_data"` - // Block height on the old chain where this transaction was confirmed. NULL if confirmation was not recorded (e.g. relay restart). - ConfirmedBlock pgtype.Int8 `json:"confirmed_block"` -} - type Notification struct { ID int64 `json:"id"` Specifier string `json:"specifier"` diff --git a/api/new_chain_flusher.go b/api/new_chain_flusher.go deleted file mode 100644 index ddef5248..00000000 --- a/api/new_chain_flusher.go +++ /dev/null @@ -1,272 +0,0 @@ -package api - -import ( - "context" - "crypto/tls" - "net" - "net/http" - "strings" - "time" - - "api.audius.co/config" - "connectrpc.com/connect" - v1 "github.com/OpenAudio/go-openaudio/pkg/api/core/v1" - corev1connect "github.com/OpenAudio/go-openaudio/pkg/api/core/v1/v1connect" - "github.com/google/uuid" - "github.com/jackc/pgx/v5/pgxpool" - "go.uber.org/zap" - "golang.org/x/net/http2" - "google.golang.org/protobuf/proto" -) - -// NewChainFlusher reads pending rows from new_chain_queue and forwards them to -// the new Core chain. On startup it marks rows covered by the backfill -// (confirmed_block < cfg.NewChainFlushFromBlock) as skipped, then sends the rest -// in id order, one at a time. Sequential processing is required to preserve -// transaction ordering across users (dev apps can act on behalf of other users, -// creating cross-user dependencies). -// -// Rows are never deleted — a forwarded row gets flushed_at set and stays put. -// The genesis migration chain is regenerated before it ships, so a row deleted -// on success would survive only on the chain being discarded. Keeping them makes -// the queue a durable log: point the flusher at the rebuilt chain, set -// NewChainFlushFromBlock to the new backfill's end height, and re-drive. -type NewChainFlusher struct { - cfg *config.Config - writePool *pgxpool.Pool - chainClient corev1connect.CoreServiceClient - logger *zap.Logger -} - -func NewNewChainFlusher(cfg *config.Config, writePool *pgxpool.Pool, logger *zap.Logger) *NewChainFlusher { - chainURL := cfg.NewChainURL - if !strings.HasPrefix(chainURL, "http://") && !strings.HasPrefix(chainURL, "https://") { - chainURL = "https://" + chainURL - } - - var httpClient *http.Client - if strings.HasPrefix(chainURL, "http://") { - // h2c: plain HTTP/2, bypasses nginx and talks directly to the gRPC port - httpClient = &http.Client{ - Transport: &http2.Transport{ - AllowHTTP: true, - DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { - return net.Dial(network, addr) - }, - }, - } - } else if cfg.NewChainInsecureSkipVerify { - httpClient = &http.Client{ - Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - }, - } - } else { - httpClient = http.DefaultClient - } - - return newFlusherWithClient(cfg, writePool, corev1connect.NewCoreServiceClient(httpClient, chainURL), logger) -} - -// newFlusherWithClient constructs a flusher with a pre-built chain client. -// Used in tests to inject a plain HTTP/1.1 client against httptest.Server. -func newFlusherWithClient(cfg *config.Config, writePool *pgxpool.Pool, client corev1connect.CoreServiceClient, logger *zap.Logger) *NewChainFlusher { - return &NewChainFlusher{ - cfg: cfg, - writePool: writePool, - chainClient: client, - logger: logger.With(zap.String("component", "new_chain_flusher")), - } -} - -// Start trims backfill-covered rows, then continuously drains new_chain_queue. -// Runs until ctx is cancelled. -func (f *NewChainFlusher) Start(ctx context.Context) { - if err := f.trimBackfillRows(ctx); err != nil { - f.logger.Error("trim failed", zap.Error(err)) - // non-fatal: continue flushing with whatever rows remain - } - - f.logger.Info("starting flush loop", - zap.String("new_chain_url", f.cfg.NewChainURL), - ) - - const batchSize = 500 - - for { - if ctx.Err() != nil { - return - } - - rows, err := f.fetchBatch(ctx, batchSize) - if err != nil { - if ctx.Err() != nil { - return - } - f.logger.Error("fetch batch failed", zap.Error(err)) - sleep(ctx, 2*time.Second) - continue - } - - if len(rows) == 0 { - // Queue is empty; pause before checking again. - sleep(ctx, 500*time.Millisecond) - continue - } - - for _, row := range rows { - if ctx.Err() != nil { - return - } - if err := f.flushRow(ctx, row); err != nil && ctx.Err() == nil { - // Do not advance past a failed row: ordering requires every tx - // to land before the next one is sent. Sleep and let the outer - // loop re-fetch the same row. - f.logger.Error("flush row failed, retrying", zap.Int64("id", row.id), zap.Error(err)) - sleep(ctx, 2*time.Second) - break - } - } - } -} - -type queueRow struct { - id int64 - txRaw []byte -} - -// fetchBatch returns the next rows to send, oldest first. -// -// NewChainFlushToBlock, when set, is a ceiling: rows confirmed above it are left -// pending. It is the mirror of NewChainFlushFromBlock and exists for the indexer -// cutover, where the old-chain indexer stops at a height L and everything -// confirmed at or below L must be on the new chain before the new indexer starts -// (ROLLOUT.md, Runbook step 12). -// -// A filter, not a stop. Enqueue is dispatched asynchronously, so confirmed_block -// is only roughly ordered by id — halting at the first row above the ceiling -// would strand a row below it, which would then flush after the boundary was -// recorded and be indexed twice. -// -// Rows with a NULL confirmed_block are held while the ceiling is set: they -// cannot be placed relative to L, and holding is the recoverable choice — -// they flush once the ceiling is lifted. Drain or inspect them before the -// cutover rather than discovering them during it. -func (f *NewChainFlusher) fetchBatch(ctx context.Context, limit int) ([]queueRow, error) { - query := `SELECT id, tx_data FROM new_chain_queue - WHERE flushed_at IS NULL - ORDER BY id LIMIT $1` - args := []any{limit} - if f.cfg.NewChainFlushToBlock > 0 { - query = `SELECT id, tx_data FROM new_chain_queue - WHERE flushed_at IS NULL - AND confirmed_block IS NOT NULL AND confirmed_block <= $2 - ORDER BY id LIMIT $1` - args = append(args, f.cfg.NewChainFlushToBlock) - } - rows, err := f.writePool.Query(ctx, query, args...) - if err != nil { - return nil, err - } - defer rows.Close() - - var result []queueRow - for rows.Next() { - var r queueRow - if err := rows.Scan(&r.id, &r.txRaw); err != nil { - return nil, err - } - result = append(result, r) - } - return result, rows.Err() -} - -func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { - var tx v1.ManageEntityLegacy - if err := proto.Unmarshal(row.txRaw, &tx); err != nil { - // Corrupt row — mark it skipped and move on rather than retrying forever. - // It stays in the table so the bad payload can still be inspected. - f.logger.Error("corrupt queue row, skipping", zap.Int64("id", row.id), zap.Error(err)) - _, _ = f.writePool.Exec(ctx, - `UPDATE new_chain_queue SET flushed_at = now(), skip_reason = 'corrupt' WHERE id = $1`, - row.id) - return nil - } - - // The chain ID tells core which chain this forward is meant for. A node on - // any other chain refuses it, so a newChainUrl pointed at the wrong chain - // stalls here instead of writing the backlog there. - req := connect.NewRequest(&v1.ForwardTransactionRequest{ - Transaction: &v1.SignedTransaction{ - RequestId: uuid.NewString(), - Transaction: &v1.SignedTransaction_ManageEntity{ - ManageEntity: &tx, - }, - }, - ChainId: f.cfg.NewChainID, - }) - - const ( - baseDelay = 500 * time.Millisecond - maxDelay = 15 * time.Second - ) - delay := baseDelay - for { - _, err := f.chainClient.ForwardTransaction(ctx, req) - if err == nil { - break - } - if !strings.Contains(err.Error(), "mempool full") { - return err - } - f.logger.Warn("mempool full, pausing flush", zap.Duration("wait", delay)) - sleep(ctx, delay) - if ctx.Err() != nil { - return ctx.Err() - } - delay *= 2 - if delay > maxDelay { - delay = maxDelay - } - } - - _, err := f.writePool.Exec(ctx, - `UPDATE new_chain_queue SET flushed_at = now() WHERE id = $1`, - row.id) - return err -} - -// trimBackfillRows marks every pending row whose confirmed_block is before the -// configured flush-from block as skipped, i.e. already covered by the genesis -// backfill. Rows with a NULL confirmed_block are left pending: NULL < $1 -// evaluates to NULL (falsy) in SQL. -// -// Only pending rows are touched, so re-running with a higher flush-from block -// after a chain regeneration marks the newly-covered rows without disturbing -// the record of what was already sent. -func (f *NewChainFlusher) trimBackfillRows(ctx context.Context) error { - if f.cfg.NewChainFlushFromBlock <= 0 { - return nil - } - tag, err := f.writePool.Exec(ctx, - `UPDATE new_chain_queue - SET flushed_at = now(), skip_reason = 'backfilled' - WHERE confirmed_block < $1 AND flushed_at IS NULL`, - f.cfg.NewChainFlushFromBlock, - ) - if err != nil { - return err - } - f.logger.Info("marked backfill-covered rows as skipped", - zap.Int64("skipped", tag.RowsAffected()), - zap.Int64("flush_from_block", f.cfg.NewChainFlushFromBlock), - ) - return nil -} - -func sleep(ctx context.Context, d time.Duration) { - select { - case <-ctx.Done(): - case <-time.After(d): - } -} diff --git a/api/new_chain_flusher_test.go b/api/new_chain_flusher_test.go deleted file mode 100644 index 296dfd88..00000000 --- a/api/new_chain_flusher_test.go +++ /dev/null @@ -1,456 +0,0 @@ -package api - -import ( - "context" - "net/http" - "net/http/httptest" - "sync/atomic" - "testing" - "time" - - "api.audius.co/config" - "api.audius.co/database" - "connectrpc.com/connect" - corev1 "github.com/OpenAudio/go-openaudio/pkg/api/core/v1" - corev1connect "github.com/OpenAudio/go-openaudio/pkg/api/core/v1/v1connect" - "github.com/stretchr/testify/require" - "go.uber.org/zap" - "google.golang.org/protobuf/proto" -) - -// mockCoreService implements only ForwardTransaction; all other methods return Unimplemented. -type mockCoreService struct { - corev1connect.UnimplementedCoreServiceHandler - received []*corev1.ManageEntityLegacy - chainIDs []string - calls atomic.Int32 -} - -func (m *mockCoreService) ForwardTransaction(_ context.Context, req *connect.Request[corev1.ForwardTransactionRequest]) (*connect.Response[corev1.ForwardTransactionResponse], error) { - m.calls.Add(1) - m.chainIDs = append(m.chainIDs, req.Msg.ChainId) - if me := req.Msg.Transaction.GetManageEntity(); me != nil { - m.received = append(m.received, me) - } - return connect.NewResponse(&corev1.ForwardTransactionResponse{}), nil -} - -func newTestFlusher(t *testing.T, cfg *config.Config) (*NewChainFlusher, *mockCoreService) { - t.Helper() - mock := &mockCoreService{} - mux := http.NewServeMux() - path, handler := corev1connect.NewCoreServiceHandler(mock) - mux.Handle(path, handler) - srv := httptest.NewServer(mux) - t.Cleanup(srv.Close) - - pool := database.CreateTestDatabase(t, "test_api") - - // Fallback for a template DB predating the new_chain_queue migrations; - // a current template already has the table and this is a no-op. - _, err := pool.Exec(context.Background(), ` - CREATE TABLE IF NOT EXISTS new_chain_queue ( - id bigserial PRIMARY KEY, - created_at timestamptz NOT NULL DEFAULT now(), - tx_data bytea NOT NULL, - confirmed_block bigint, - flushed_at timestamptz, - skip_reason text - ) - `) - require.NoError(t, err) - - cfg.NewChainURL = srv.URL - - logger, _ := zap.NewDevelopment() - client := corev1connect.NewCoreServiceClient(http.DefaultClient, srv.URL) - f := newFlusherWithClient(cfg, pool, client, logger) - return f, mock -} - -func sampleTx(entityID int64) *corev1.ManageEntityLegacy { - return &corev1.ManageEntityLegacy{ - UserId: 1, - EntityType: "Track", - EntityId: entityID, - Action: "Create", - Metadata: "{}", - Nonce: "0xdeadbeef", - } -} - -func insertQueueRow(t *testing.T, f *NewChainFlusher, tx *corev1.ManageEntityLegacy, confirmedBlock *int64) { - t.Helper() - b, err := proto.Marshal(tx) - require.NoError(t, err) - _, err = f.writePool.Exec(context.Background(), - `INSERT INTO new_chain_queue (tx_data, confirmed_block) VALUES ($1, $2)`, - b, confirmedBlock, - ) - require.NoError(t, err) -} - -// pendingDepth counts rows the flusher still has to send. Rows are never -// deleted, so this is the count of unflushed rows rather than the table size. -func pendingDepth(t *testing.T, f *NewChainFlusher) int { - t.Helper() - var n int - err := f.writePool.QueryRow(context.Background(), - `SELECT count(*) FROM new_chain_queue WHERE flushed_at IS NULL`).Scan(&n) - require.NoError(t, err) - return n -} - -// totalDepth counts every row, flushed or not. -func totalDepth(t *testing.T, f *NewChainFlusher) int { - t.Helper() - var n int - err := f.writePool.QueryRow(context.Background(), `SELECT count(*) FROM new_chain_queue`).Scan(&n) - require.NoError(t, err) - return n -} - -// skipReasons returns the skip_reason of every non-pending row, in id order, -// with NULL (i.e. genuinely forwarded) rendered as "sent". -func skipReasons(t *testing.T, f *NewChainFlusher) []string { - t.Helper() - rows, err := f.writePool.Query(context.Background(), - `SELECT coalesce(skip_reason, 'sent') FROM new_chain_queue - WHERE flushed_at IS NOT NULL ORDER BY id`) - require.NoError(t, err) - defer rows.Close() - var out []string - for rows.Next() { - var r string - require.NoError(t, rows.Scan(&r)) - out = append(out, r) - } - require.NoError(t, rows.Err()) - return out -} - -// TestEnqueueForNewChain verifies that enqueueForNewChain inserts a row with the -// correct tx_data and confirmed_block. -func TestEnqueueForNewChain(t *testing.T) { - app := emptyTestApp(t) - - // Fallback for a template DB predating the new_chain_queue migrations. - _, err := app.writePool.Exec(context.Background(), ` - CREATE TABLE IF NOT EXISTS new_chain_queue ( - id bigserial PRIMARY KEY, - created_at timestamptz NOT NULL DEFAULT now(), - tx_data bytea NOT NULL, - confirmed_block bigint, - flushed_at timestamptz, - skip_reason text - ) - `) - require.NoError(t, err) - - tx := sampleTx(42) - app.enqueueForNewChain(tx, 999) - - // enqueueForNewChain is fire-and-forget; give it a moment to write. - require.Eventually(t, func() bool { - var n int - app.writePool.QueryRow(context.Background(), `SELECT count(*) FROM new_chain_queue`).Scan(&n) //nolint:errcheck - return n == 1 - }, 2*time.Second, 50*time.Millisecond) - - var txData []byte - var confirmedBlock int64 - err = app.writePool.QueryRow(context.Background(), - `SELECT tx_data, confirmed_block FROM new_chain_queue LIMIT 1`, - ).Scan(&txData, &confirmedBlock) - require.NoError(t, err) - require.Equal(t, int64(999), confirmedBlock) - - var got corev1.ManageEntityLegacy - require.NoError(t, proto.Unmarshal(txData, &got)) - require.Equal(t, int64(42), got.EntityId) - require.Equal(t, "Track", got.EntityType) -} - -// TestNewChainFlusherTrim verifies that rows with confirmed_block < FlushFromBlock -// are marked skipped on startup rather than deleted, and rows at or above the -// threshold stay pending. -func TestNewChainFlusherTrim(t *testing.T) { - cfg := &config.Config{NewChainFlushFromBlock: 100} - f, _ := newTestFlusher(t, cfg) - - block50 := int64(50) - block99 := int64(99) - block100 := int64(100) - block200 := int64(200) - insertQueueRow(t, f, sampleTx(1), &block50) // should be trimmed - insertQueueRow(t, f, sampleTx(2), &block99) // should be trimmed - insertQueueRow(t, f, sampleTx(3), &block100) // kept (boundary) - insertQueueRow(t, f, sampleTx(4), &block200) // kept - insertQueueRow(t, f, sampleTx(5), nil) // NULL confirmed_block — kept - - require.Equal(t, 5, pendingDepth(t, f)) - - err := f.trimBackfillRows(context.Background()) - require.NoError(t, err) - - require.Equal(t, 3, pendingDepth(t, f), "two backfill-covered rows should no longer be pending") - require.Equal(t, 5, totalDepth(t, f), "trim must not delete rows") - require.Equal(t, []string{"backfilled", "backfilled"}, skipReasons(t, f)) - - // Verify the entity IDs still pending. - rows, err := f.writePool.Query(context.Background(), - `SELECT (tx_data) FROM new_chain_queue WHERE flushed_at IS NULL ORDER BY id`, - ) - require.NoError(t, err) - defer rows.Close() - - var ids []int64 - for rows.Next() { - var b []byte - require.NoError(t, rows.Scan(&b)) - var me corev1.ManageEntityLegacy - require.NoError(t, proto.Unmarshal(b, &me)) - ids = append(ids, me.EntityId) - } - require.Equal(t, []int64{3, 4, 5}, ids) -} - -// TestNewChainFlusherSends verifies that the flusher forwards all queued rows to -// the new chain and marks them flushed — retaining them — on success. -func TestNewChainFlusherSends(t *testing.T) { - cfg := &config.Config{NewChainID: "audius-mainnet-beta"} // no trim - f, mock := newTestFlusher(t, cfg) - - block10 := int64(10) - block20 := int64(20) - block30 := int64(30) - insertQueueRow(t, f, sampleTx(1), &block10) - insertQueueRow(t, f, sampleTx(2), &block20) - insertQueueRow(t, f, sampleTx(3), &block30) - - require.Equal(t, 3, pendingDepth(t, f)) - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - go f.Start(ctx) - - require.Eventually(t, func() bool { - return mock.calls.Load() == 3 - }, 5*time.Second, 50*time.Millisecond, "expected 3 ForwardTransaction calls") - - require.Eventually(t, func() bool { - return pendingDepth(t, f) == 0 - }, 5*time.Second, 50*time.Millisecond, "expected queue to drain") - - // Forwarded rows are retained so they can be re-driven onto a rebuilt chain. - require.Equal(t, 3, totalDepth(t, f), "flushed rows must not be deleted") - require.Equal(t, []string{"sent", "sent", "sent"}, skipReasons(t, f)) - - // Verify all three entity IDs were forwarded. - receivedIDs := make([]int64, len(mock.received)) - for i, me := range mock.received { - receivedIDs[i] = me.EntityId - } - require.ElementsMatch(t, []int64{1, 2, 3}, receivedIDs) - - // Every forward names the new chain, or core refuses it. - require.Equal(t, []string{"audius-mainnet-beta", "audius-mainnet-beta", "audius-mainnet-beta"}, mock.chainIDs) -} - -// TestNewChainFlusherTrimThenSend verifies the combined trim+flush flow: -// rows before the cutoff block are dropped; rows after are forwarded. -func TestNewChainFlusherTrimThenSend(t *testing.T) { - cfg := &config.Config{NewChainFlushFromBlock: 50} - f, mock := newTestFlusher(t, cfg) - - block10 := int64(10) // pre-backfill — trimmed - block49 := int64(49) // pre-backfill — trimmed - block50 := int64(50) // post-backfill — flushed - block100 := int64(100) // post-backfill — flushed - insertQueueRow(t, f, sampleTx(1), &block10) - insertQueueRow(t, f, sampleTx(2), &block49) - insertQueueRow(t, f, sampleTx(3), &block50) - insertQueueRow(t, f, sampleTx(4), &block100) - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - go f.Start(ctx) - - require.Eventually(t, func() bool { - return mock.calls.Load() == 2 - }, 5*time.Second, 50*time.Millisecond, "expected 2 ForwardTransaction calls (post-trim)") - - require.Eventually(t, func() bool { - return pendingDepth(t, f) == 0 - }, 5*time.Second, 50*time.Millisecond, "expected queue to drain") - - // Two skipped by the trim, two genuinely sent — all four retained. - require.Equal(t, 4, totalDepth(t, f)) - require.Equal(t, []string{"backfilled", "backfilled", "sent", "sent"}, skipReasons(t, f)) - - receivedIDs := make([]int64, len(mock.received)) - for i, me := range mock.received { - receivedIDs[i] = me.EntityId - } - require.ElementsMatch(t, []int64{3, 4}, receivedIDs) -} - -// TestNewChainFlusherRedriveAfterRegeneration is the reason rows are retained -// rather than deleted. The genesis chain is regenerated before it ships, so -// everything already forwarded has to be replayable onto the rebuilt chain. -// -// Simulates that: drain the queue against chain v1, then clear flushed_at (what -// an operator does when repointing at the rebuilt chain) and confirm the same -// transactions are forwarded again. With delete-on-success there would be -// nothing left to re-drive. -func TestNewChainFlusherRedriveAfterRegeneration(t *testing.T) { - cfg := &config.Config{} - f, mock := newTestFlusher(t, cfg) - - block10 := int64(10) - block20 := int64(20) - insertQueueRow(t, f, sampleTx(1), &block10) - insertQueueRow(t, f, sampleTx(2), &block20) - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - go f.Start(ctx) - - require.Eventually(t, func() bool { - return mock.calls.Load() == 2 - }, 5*time.Second, 50*time.Millisecond, "expected both rows forwarded to chain v1") - require.Eventually(t, func() bool { - return pendingDepth(t, f) == 0 - }, 5*time.Second, 50*time.Millisecond) - cancel() - - // The rows are still here — that is the point. - require.Equal(t, 2, totalDepth(t, f)) - - // Repoint at the rebuilt chain: everything becomes pending again. - _, err := f.writePool.Exec(context.Background(), - `UPDATE new_chain_queue SET flushed_at = NULL, skip_reason = NULL`) - require.NoError(t, err) - require.Equal(t, 2, pendingDepth(t, f)) - - ctx2, cancel2 := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel2() - go f.Start(ctx2) - - require.Eventually(t, func() bool { - return mock.calls.Load() == 4 - }, 5*time.Second, 50*time.Millisecond, "expected both rows forwarded again after re-drive") - require.Eventually(t, func() bool { - return pendingDepth(t, f) == 0 - }, 5*time.Second, 50*time.Millisecond) -} - -// The ceiling is the mirror of NewChainFlushFromBlock: rows confirmed above it -// stay pending. It exists so everything confirmed at or below the indexer's stop -// height L can be landed on the new chain before the new indexer starts, without -// letting anything above L across the boundary first. -func TestNewChainFlusherCeilingHoldsRowsAboveIt(t *testing.T) { - cfg := &config.Config{NewChainFlushToBlock: 100} - f, _ := newTestFlusher(t, cfg) - - block50 := int64(50) - block100 := int64(100) - block101 := int64(101) - block200 := int64(200) - insertQueueRow(t, f, sampleTx(1), &block50) // below ceiling — eligible - insertQueueRow(t, f, sampleTx(2), &block100) // at ceiling — eligible, inclusive - insertQueueRow(t, f, sampleTx(3), &block101) // above — held - insertQueueRow(t, f, sampleTx(4), &block200) // above — held - - batch, err := f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Len(t, batch, 2, "only rows confirmed at or below the ceiling are eligible") - require.Equal(t, []int64{1, 2}, entityIDsOf(t, batch)) -} - -// A row below the ceiling sitting behind one above it must still be returned. -// Enqueue is fire-and-forget, so confirmed_block is only roughly ordered by id; -// a halt-on-first-above-ceiling would strand this row, and it would then flush -// after the boundary was recorded and be indexed a second time. -func TestNewChainFlusherCeilingFiltersRatherThanHalts(t *testing.T) { - cfg := &config.Config{NewChainFlushToBlock: 100} - f, _ := newTestFlusher(t, cfg) - - above := int64(200) - below := int64(50) - insertQueueRow(t, f, sampleTx(1), &above) // lower id, above the ceiling - insertQueueRow(t, f, sampleTx(2), &below) // higher id, below it - - batch, err := f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Equal(t, []int64{2}, entityIDsOf(t, batch), - "the row below the ceiling must be reachable past one above it") -} - -// NULL confirmed_block cannot be placed relative to the ceiling, so it is held -// while one is set and flushes once it is lifted. Holding is the recoverable -// choice: sending it early could double-index, dropping it would lose the write. -func TestNewChainFlusherCeilingHoldsNullConfirmedBlock(t *testing.T) { - cfg := &config.Config{NewChainFlushToBlock: 100} - f, _ := newTestFlusher(t, cfg) - - insertQueueRow(t, f, sampleTx(1), nil) - - batch, err := f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Empty(t, batch, "NULL confirmed_block is held while a ceiling is set") - - // Lifting the ceiling releases it. - f.cfg.NewChainFlushToBlock = 0 - batch, err = f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Equal(t, []int64{1}, entityIDsOf(t, batch)) -} - -// With no ceiling configured the flusher is unbounded, as in normal operation. -func TestNewChainFlusherNoCeilingSendsEverything(t *testing.T) { - cfg := &config.Config{} - f, _ := newTestFlusher(t, cfg) - - block50 := int64(50) - insertQueueRow(t, f, sampleTx(1), &block50) - insertQueueRow(t, f, sampleTx(2), nil) - - batch, err := f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Equal(t, []int64{1, 2}, entityIDsOf(t, batch)) -} - -func entityIDsOf(t *testing.T, batch []queueRow) []int64 { - t.Helper() - var ids []int64 - for _, r := range batch { - var me corev1.ManageEntityLegacy - require.NoError(t, proto.Unmarshal(r.txRaw, &me)) - ids = append(ids, me.EntityId) - } - return ids -} - -// The ceiling narrows which pending rows are eligible; it must never resurrect -// a row that was already forwarded or skipped. Both filters apply together. -func TestNewChainFlusherCeilingSkipsFlushedRows(t *testing.T) { - cfg := &config.Config{NewChainFlushToBlock: 100} - f, _ := newTestFlusher(t, cfg) - - block50 := int64(50) - block60 := int64(60) - insertQueueRow(t, f, sampleTx(1), &block50) - insertQueueRow(t, f, sampleTx(2), &block60) - - _, err := f.writePool.Exec(context.Background(), - `UPDATE new_chain_queue SET flushed_at = now() WHERE confirmed_block = 50`) - require.NoError(t, err) - - batch, err := f.fetchBatch(context.Background(), 100) - require.NoError(t, err) - require.Equal(t, []int64{2}, entityIDsOf(t, batch), - "a flushed row below the ceiling must not be sent again") -} diff --git a/api/relay.go b/api/relay.go index c702a8a1..04b2025e 100644 --- a/api/relay.go +++ b/api/relay.go @@ -19,7 +19,6 @@ import ( "github.com/ethereum/go-ethereum/common/hexutil" "github.com/gofiber/fiber/v2" "go.uber.org/zap" - "google.golang.org/protobuf/proto" ) const ( @@ -287,12 +286,6 @@ func (app *ApiServer) handleRelay(ctx context.Context, logger *zap.Logger, decod msg := res.Msg.Transaction endpointLogger.Info("transaction confirmed", zap.String("hash", msg.GetHash())) - // Enqueue for new chain if dual-write is enabled. Fire-and-forget: - // a queue failure must never fail the relay response to the client. - if app.config.NewChainQueueEnabled && app.writePool != nil { - go app.enqueueForNewChain(decodedTx, msg.GetHeight()) - } - return msg, nil } @@ -300,23 +293,6 @@ func (app *ApiServer) handleRelay(ctx context.Context, logger *zap.Logger, decod return nil, fmt.Errorf("all endpoints failed, last error: %w", lastErr) } -// enqueueForNewChain serializes a confirmed ManageEntityLegacy transaction and -// inserts it into new_chain_queue. Called asynchronously; logs errors but never panics. -func (app *ApiServer) enqueueForNewChain(tx *v1.ManageEntityLegacy, confirmedBlock int64) { - b, err := proto.Marshal(tx) - if err != nil { - app.logger.Warn("new_chain_queue: marshal failed", zap.Error(err)) - return - } - _, err = app.writePool.Exec(context.Background(), - `INSERT INTO new_chain_queue (tx_data, confirmed_block) VALUES ($1, $2)`, - b, confirmedBlock, - ) - if err != nil { - app.logger.Warn("new_chain_queue: insert failed", zap.Error(err)) - } -} - func transactionToReceipt(tx *v1.Transaction, wallet string) map[string]interface{} { return map[string]interface{}{ "transactionHash": tx.GetHash(), diff --git a/api/server.go b/api/server.go index 74d9b386..9acb6b34 100644 --- a/api/server.go +++ b/api/server.go @@ -287,15 +287,6 @@ func NewApiServer(config config.Config) *ApiServer { panic(err) } - var newChainFlusher *NewChainFlusher - if config.NewChainFlushEnabled && config.NewChainURL != "" && writePool != nil { - if config.NewChainID == "" { - logger.Error("new chain flusher not started: newChainId is not set, and the new chain refuses forwards without a chain ID") - } else { - newChainFlusher = NewNewChainFlusher(&config, writePool, logger) - } - } - app := &ApiServer{ App: fiber.New(fiber.Config{ JSONEncoder: json.Marshal, @@ -340,7 +331,6 @@ func NewApiServer(config config.Config) *ApiServer { rateLimitMiddleware: rateLimitMiddleware, solanaRpcClient: solanaRpc, meteoraDbcClient: meteoraDbcClient, - newChainFlusher: newChainFlusher, } // Set up a custom decoder for HashIds so they can be parsed in lists @@ -908,7 +898,6 @@ type ApiServer struct { meteoraDbcClient *meteora_dbc.Client validators *Nodes openAudioPool *OpenAudioPool - newChainFlusher *NewChainFlusher } // requestLogger returns app.logger annotated with the current request_id (set @@ -999,14 +988,6 @@ func (app *ApiServer) Serve() { app.logger.Info("Started validators poller") }() - if app.newChainFlusher != nil { - go func() { - app.logger.Info("Starting new chain flusher") - app.newChainFlusher.Start(ctx) - app.logger.Info("New chain flusher stopped") - }() - } - // Bind to both ipv4 and ipv6 listener, err := net.Listen("tcp", "[::]:1323") if err != nil { diff --git a/config/config.go b/config/config.go index 37c21373..0db39d09 100644 --- a/config/config.go +++ b/config/config.go @@ -81,18 +81,6 @@ type Config struct { // prioritization (the list reduces to open-before-ended). FeaturedAudienceUserID int32 - // Genesis migration dual-write queue. - // NewChainURL is the bootstrap chain gRPC endpoint (e.g. http://bootstrap-node:50051). - // NewChainQueueEnabled turns on enqueuing of relayed txs for the new chain. - // NewChainFlushEnabled turns on the background flusher goroutine. - // NewChainFlushFromBlock, when set, causes the flusher to delete all queued rows with - // confirmed_block < NewChainFlushFromBlock before sending — trimming rows already - // covered by the backfill. - // NewChainInsecureSkipVerify disables TLS verification for the new chain endpoint (e.g. localstack). - // NewChainID is the chain the flusher stamps on every ForwardTransaction. - // Core refuses a forward whose chain ID differs from its own, and on the - // new chain also one with none, so the flusher does not start without it. - // Defaults per environment; newChainId overrides. // PlayRoutingHosts, when non-empty, are tried first when resolving a track // stream URL. Plays are recorded by whichever node serves the audio, never // through the relay, so this is what decides which chain a play lands on. @@ -108,14 +96,6 @@ type Config struct { // cutover. See cmd/genesis-writer/ROLLOUT.md, Runbook steps 5 and 13. PlayRoutingHosts []string - NewChainURL string - NewChainID string - NewChainQueueEnabled bool - NewChainFlushEnabled bool - NewChainFlushFromBlock int64 - NewChainFlushToBlock int64 - NewChainInsecureSkipVerify bool - // Indexer cutover bounds. Both default to 0, meaning "unset" — the ETL // treats a zero start as "resume from where you left off" and a zero end as // "never stop", which is normal operation. @@ -212,7 +192,6 @@ func init() { Cfg.Rewards = core_config.MakeRewards(core_config.DevClaimAuthorities, core_config.DevRewardExtensions) Cfg.AudiusdURL = "https://node1.oap.devnet" Cfg.ChainId = "openaudio-devnet" - Cfg.NewChainID = "openaudio-devnet" Cfg.SolanaIndexerWorkers = 1 Cfg.DeadNodes = []string{} Cfg.BlacklistedNodes = []string{} @@ -312,7 +291,6 @@ func init() { Cfg.Rewards = core_config.MakeRewards(core_config.ProdClaimAuthorities, core_config.ProdRewardExtensions) Cfg.AudiusdURL = "rpc.audius.co" Cfg.ChainId = "audius-mainnet-alpha-beta" - Cfg.NewChainID = "audius-mainnet-beta" Cfg.AudiusdChainID = core_config.ProdAcdcChainID Cfg.AudiusdEntityManagerAddress = core_config.ProdAcdcAddress Cfg.AudiusAppUrl = "https://audius.co" @@ -403,17 +381,6 @@ func init() { Cfg.FeaturedAudienceUserID = int32(parsed) } - // Genesis migration dual-write queue - Cfg.NewChainURL = os.Getenv("newChainUrl") - if v := os.Getenv("newChainId"); v != "" { - Cfg.NewChainID = v - } - Cfg.NewChainQueueEnabled = os.Getenv("newChainQueueEnabled") == "true" - Cfg.NewChainFlushEnabled = os.Getenv("newChainFlushEnabled") == "true" - Cfg.NewChainInsecureSkipVerify = os.Getenv("newChainInsecureSkipVerify") == "true" - Cfg.NewChainFlushFromBlock = mustParseInt64Env("newChainFlushFromBlock") - Cfg.NewChainFlushToBlock = mustParseInt64Env("newChainFlushToBlock") - // Indexer cutover bounds (see the struct fields). Cfg.EtlStartingBlockHeight = mustParseInt64Env("etlStartingBlockHeight") Cfg.EtlEndingBlockHeight = mustParseInt64Env("etlEndingBlockHeight") diff --git a/ddl/migrations/0243_drop_new_chain_queue.sql b/ddl/migrations/0243_drop_new_chain_queue.sql new file mode 100644 index 00000000..7c8245a9 --- /dev/null +++ b/ddl/migrations/0243_drop_new_chain_queue.sql @@ -0,0 +1,7 @@ +-- Retire new_chain_queue. The API no longer enqueues or flushes writes for the +-- new chain; a standalone mirror in go-openaudio replays the old chain instead. +BEGIN; + +DROP TABLE IF EXISTS new_chain_queue; + +COMMIT; diff --git a/sql/01_schema.sql b/sql/01_schema.sql index 28b6a124..8c81f1fb 100644 --- a/sql/01_schema.sql +++ b/sql/01_schema.sql @@ -9117,74 +9117,6 @@ CREATE MATERIALIZED VIEW public.mv_dashboard_transaction_types AS WITH NO DATA; --- --- Name: new_chain_queue; Type: TABLE; Schema: public; Owner: - --- - -CREATE TABLE public.new_chain_queue ( - id bigint NOT NULL, - created_at timestamp with time zone DEFAULT now() NOT NULL, - tx_data bytea NOT NULL, - confirmed_block bigint, - flushed_at timestamp with time zone, - skip_reason text -); - - --- --- Name: TABLE new_chain_queue; Type: COMMENT; Schema: public; Owner: - --- - -COMMENT ON TABLE public.new_chain_queue IS 'Queue of ManageEntity transactions to be forwarded to the new Core chain (audius-mainnet-v2) during genesis migration.'; - - --- --- Name: COLUMN new_chain_queue.tx_data; Type: COMMENT; Schema: public; Owner: - --- - -COMMENT ON COLUMN public.new_chain_queue.tx_data IS 'Protobuf-serialized ManageEntityLegacy message.'; - - --- --- Name: COLUMN new_chain_queue.confirmed_block; Type: COMMENT; Schema: public; Owner: - --- - -COMMENT ON COLUMN public.new_chain_queue.confirmed_block IS 'Block height on the old chain where this transaction was confirmed. NULL if confirmation was not recorded (e.g. relay restart).'; - - --- --- Name: COLUMN new_chain_queue.flushed_at; Type: COMMENT; Schema: public; Owner: - --- - -COMMENT ON COLUMN public.new_chain_queue.flushed_at IS 'When this row was forwarded to the new chain, or when it was marked skipped. NULL means pending.'; - - --- --- Name: COLUMN new_chain_queue.skip_reason; Type: COMMENT; Schema: public; Owner: - --- - -COMMENT ON COLUMN public.new_chain_queue.skip_reason IS 'Set when flushed_at was recorded without actually forwarding: ''backfilled'' (covered by the genesis backfill) or ''corrupt'' (tx_data failed to unmarshal).'; - - --- --- Name: new_chain_queue_id_seq; Type: SEQUENCE; Schema: public; Owner: - --- - -CREATE SEQUENCE public.new_chain_queue_id_seq - START WITH 1 - INCREMENT BY 1 - NO MINVALUE - NO MAXVALUE - CACHE 1; - - --- --- Name: new_chain_queue_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: - --- - -ALTER SEQUENCE public.new_chain_queue_id_seq OWNED BY public.new_chain_queue.id; - - -- -- Name: notification; Type: TABLE; Schema: public; Owner: - -- @@ -11791,13 +11723,6 @@ ALTER TABLE ONLY public.etl_validator_registrations ALTER COLUMN id SET DEFAULT ALTER TABLE ONLY public.etl_validators ALTER COLUMN id SET DEFAULT nextval('public.etl_validators_id_seq'::regclass); --- --- Name: new_chain_queue id; Type: DEFAULT; Schema: public; Owner: - --- - -ALTER TABLE ONLY public.new_chain_queue ALTER COLUMN id SET DEFAULT nextval('public.new_chain_queue_id_seq'::regclass); - - -- -- Name: notification id; Type: DEFAULT; Schema: public; Owner: - -- @@ -12548,14 +12473,6 @@ ALTER TABLE ONLY public.muted_users ADD CONSTRAINT muted_users_pkey PRIMARY KEY (muted_user_id, user_id); --- --- Name: new_chain_queue new_chain_queue_pkey; Type: CONSTRAINT; Schema: public; Owner: - --- - -ALTER TABLE ONLY public.new_chain_queue - ADD CONSTRAINT new_chain_queue_pkey PRIMARY KEY (id); - - -- -- Name: notification_campaign_push_open notification_campaign_push_open_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- @@ -14702,20 +14619,6 @@ CREATE INDEX mv_dashboard_transaction_stats_24h_idx ON public.mv_dashboard_trans CREATE INDEX mv_dashboard_transaction_types_idx ON public.mv_dashboard_transaction_types USING btree (tx_type, transaction_count); --- --- Name: new_chain_queue_confirmed_block_idx; Type: INDEX; Schema: public; Owner: - --- - -CREATE INDEX new_chain_queue_confirmed_block_idx ON public.new_chain_queue USING btree (confirmed_block); - - --- --- Name: new_chain_queue_pending_idx; Type: INDEX; Schema: public; Owner: - --- - -CREATE INDEX new_chain_queue_pending_idx ON public.new_chain_queue USING btree (id) WHERE (flushed_at IS NULL); - - -- -- Name: notification_multi_recipient_user_ids_idx; Type: INDEX; Schema: public; Owner: - --