From 4fcc85a94e29f2554e0cc962a9f8b0fb3ba715a0 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:37:29 -0700 Subject: [PATCH 1/2] fix(api): retain flushed new-chain queue rows instead of deleting them The flusher deleted each row after forwarding it. That is unrecoverable, and the genesis migration chain is regenerated before it ships -- the validator key is baked into every block header, so it has to be one the bootstrap node holds, which means rebuilding the chain. Anything already forwarded and deleted would survive only on the chain being discarded. Mark rows instead. A forwarded row gets flushed_at set and stays put, so the queue is a durable log: repoint the flusher at the rebuilt chain, set NewChainFlushFromBlock to the new backfill's end height, and re-drive. Replay is idempotent -- the same signed transactions are rejected as duplicates by a chain that already has them and accepted by a rebuilt one. The two non-forwarding paths are marked rather than deleted for the same reason, and record why in skip_reason: 'backfilled' for rows the genesis backfill covers, 'corrupt' for payloads that fail to unmarshal (previously deleted outright, taking the evidence with them). trimBackfillRows now only touches pending rows, so re-running it with a higher flush-from block after a regeneration does not disturb the record of what was already sent. Reads are unaffected in the steady state: the flusher only ever selects pending rows, and a partial index on (id) WHERE flushed_at IS NULL keeps retained rows off the hot path as the table grows. Retention is now an explicit operator decision rather than a side effect of flushing. Co-Authored-By: Claude Opus 5 --- api/new_chain_flusher.go | 53 ++++--- api/new_chain_flusher_test.go | 133 +++++++++++++++--- .../0238_new_chain_queue_cursor.sql | 22 +++ 3 files changed, 174 insertions(+), 34 deletions(-) create mode 100644 ddl/migrations/0238_new_chain_queue_cursor.sql diff --git a/api/new_chain_flusher.go b/api/new_chain_flusher.go index 37b1de77..1ecc1647 100644 --- a/api/new_chain_flusher.go +++ b/api/new_chain_flusher.go @@ -19,11 +19,18 @@ import ( "google.golang.org/protobuf/proto" ) -// NewChainFlusher reads rows from new_chain_queue and forwards them to the new -// Core chain. On startup it deletes rows covered by the backfill (confirmed_block -// < cfg.NewChainFlushFromBlock), 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). +// 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 @@ -130,7 +137,9 @@ type queueRow struct { func (f *NewChainFlusher) fetchBatch(ctx context.Context, limit int) ([]queueRow, error) { rows, err := f.writePool.Query(ctx, - `SELECT id, tx_data FROM new_chain_queue ORDER BY id LIMIT $1`, + `SELECT id, tx_data FROM new_chain_queue + WHERE flushed_at IS NULL + ORDER BY id LIMIT $1`, limit, ) if err != nil { @@ -152,9 +161,12 @@ func (f *NewChainFlusher) fetchBatch(ctx context.Context, limit int) ([]queueRow 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 — delete it and move on rather than retrying forever. - f.logger.Error("corrupt queue row, deleting", zap.Int64("id", row.id), zap.Error(err)) - _, _ = f.writePool.Exec(ctx, `DELETE FROM new_chain_queue WHERE id = $1`, row.id) + // 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 } @@ -191,26 +203,35 @@ func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { } } - _, err := f.writePool.Exec(ctx, `DELETE FROM new_chain_queue WHERE id = $1`, row.id) + _, err := f.writePool.Exec(ctx, + `UPDATE new_chain_queue SET flushed_at = now() WHERE id = $1`, + row.id) return err } -// trimBackfillRows deletes all queue rows whose confirmed_block is before the -// configured flush-from block, i.e. rows already covered by the genesis backfill. -// Rows with a NULL confirmed_block are kept: NULL < $1 evaluates to NULL (falsy) in SQL. +// 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, - `DELETE FROM new_chain_queue WHERE confirmed_block < $1`, + `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("trimmed backfill-covered rows", - zap.Int64("deleted", tag.RowsAffected()), + f.logger.Info("marked backfill-covered rows as skipped", + zap.Int64("skipped", tag.RowsAffected()), zap.Int64("flush_from_block", f.cfg.NewChainFlushFromBlock), ) return nil diff --git a/api/new_chain_flusher_test.go b/api/new_chain_flusher_test.go index 674f01ce..c2f68299 100644 --- a/api/new_chain_flusher_test.go +++ b/api/new_chain_flusher_test.go @@ -44,13 +44,16 @@ func newTestFlusher(t *testing.T, cfg *config.Config) (*NewChainFlusher, *mockCo pool := database.CreateTestDatabase(t, "test_api") - // Create the new_chain_queue table (not in template DB yet). + // 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 + confirmed_block bigint, + flushed_at timestamptz, + skip_reason text ) `) require.NoError(t, err) @@ -85,7 +88,19 @@ func insertQueueRow(t *testing.T, f *NewChainFlusher, tx *corev1.ManageEntityLeg require.NoError(t, err) } -func queueDepth(t *testing.T, f *NewChainFlusher) int { +// 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) @@ -93,18 +108,39 @@ func queueDepth(t *testing.T, f *NewChainFlusher) int { 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) - // Add new_chain_queue to the test DB. + // 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 + confirmed_block bigint, + flushed_at timestamptz, + skip_reason text ) `) require.NoError(t, err) @@ -134,7 +170,8 @@ func TestEnqueueForNewChain(t *testing.T) { } // TestNewChainFlusherTrim verifies that rows with confirmed_block < FlushFromBlock -// are deleted on startup, and rows at or above the threshold are kept. +// 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) @@ -147,18 +184,20 @@ func TestNewChainFlusherTrim(t *testing.T) { 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 + insertQueueRow(t, f, sampleTx(5), nil) // NULL confirmed_block — kept - require.Equal(t, 5, queueDepth(t, f)) + require.Equal(t, 5, pendingDepth(t, f)) err := f.trimBackfillRows(context.Background()) require.NoError(t, err) - require.Equal(t, 3, queueDepth(t, f)) + 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 surviving entity IDs. + // Verify the entity IDs still pending. rows, err := f.writePool.Query(context.Background(), - `SELECT (tx_data) FROM new_chain_queue ORDER BY id`, + `SELECT (tx_data) FROM new_chain_queue WHERE flushed_at IS NULL ORDER BY id`, ) require.NoError(t, err) defer rows.Close() @@ -175,7 +214,7 @@ func TestNewChainFlusherTrim(t *testing.T) { } // TestNewChainFlusherSends verifies that the flusher forwards all queued rows to -// the new chain and deletes them on success. +// the new chain and marks them flushed — retaining them — on success. func TestNewChainFlusherSends(t *testing.T) { cfg := &config.Config{} // no trim f, mock := newTestFlusher(t, cfg) @@ -187,7 +226,7 @@ func TestNewChainFlusherSends(t *testing.T) { insertQueueRow(t, f, sampleTx(2), &block20) insertQueueRow(t, f, sampleTx(3), &block30) - require.Equal(t, 3, queueDepth(t, f)) + require.Equal(t, 3, pendingDepth(t, f)) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -199,9 +238,13 @@ func TestNewChainFlusherSends(t *testing.T) { }, 5*time.Second, 50*time.Millisecond, "expected 3 ForwardTransaction calls") require.Eventually(t, func() bool { - return queueDepth(t, f) == 0 + 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 { @@ -216,9 +259,9 @@ 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 + 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) @@ -235,12 +278,66 @@ func TestNewChainFlusherTrimThenSend(t *testing.T) { }, 5*time.Second, 50*time.Millisecond, "expected 2 ForwardTransaction calls (post-trim)") require.Eventually(t, func() bool { - return queueDepth(t, f) == 0 + 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) +} diff --git a/ddl/migrations/0238_new_chain_queue_cursor.sql b/ddl/migrations/0238_new_chain_queue_cursor.sql new file mode 100644 index 00000000..b136c958 --- /dev/null +++ b/ddl/migrations/0238_new_chain_queue_cursor.sql @@ -0,0 +1,22 @@ +BEGIN; + +-- Mark queue rows as flushed instead of deleting them. +-- +-- The genesis migration chain gets regenerated before it ships (the validator +-- key baked into every block header has to be one the bootstrap node holds), so +-- any row deleted after a successful forward would survive only on a chain that +-- is about to be discarded. Retaining the rows makes new_chain_queue a durable +-- log that can be re-driven onto the rebuilt chain. +ALTER TABLE new_chain_queue ADD COLUMN IF NOT EXISTS flushed_at timestamptz; +ALTER TABLE new_chain_queue ADD COLUMN IF NOT EXISTS skip_reason text; + +COMMENT ON COLUMN new_chain_queue.flushed_at IS 'When this row was forwarded to the new chain, or when it was marked skipped. NULL means pending.'; +COMMENT ON COLUMN 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).'; + +-- The flusher only ever reads pending rows, so index those alone. The retained +-- flushed rows stay out of the index and off the hot path no matter how large +-- the table grows. +CREATE INDEX IF NOT EXISTS new_chain_queue_pending_idx + ON new_chain_queue (id) WHERE flushed_at IS NULL; + +COMMIT; From affb78a4a82bb18c95943842ebed6bf5de24b571 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:27:20 -0700 Subject: [PATCH 2/2] chore(api): regenerate the schema dump for the new_chain_queue migration 0238 added flushed_at and skip_reason but sql/01_schema.sql was never regenerated, and that dump is what the test database is built from. Every test therefore ran the new flusher against a table without the columns it writes, which is why CI failed. Produced by make test-schema, so the migration tracker moves with it. --- sql/01_schema.sql | 41 ++++++++++++++++++++++++++++++++++-- sql/03_migration_tracker.sql | 7 ++++-- 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/sql/01_schema.sql b/sql/01_schema.sql index 93751524..36d6064b 100644 --- a/sql/01_schema.sql +++ b/sql/01_schema.sql @@ -9116,7 +9116,9 @@ 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 + confirmed_block bigint, + flushed_at timestamp with time zone, + skip_reason text ); @@ -9141,6 +9143,20 @@ COMMENT ON COLUMN public.new_chain_queue.tx_data IS 'Protobuf-serialized ManageE 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: - -- @@ -14517,6 +14533,13 @@ CREATE INDEX ix_reactions_reacted_to_reaction_type ON public.reactions USING btr CREATE INDEX ix_subscriptions_blocknumber ON public.subscriptions USING btree (blocknumber); +-- +-- Name: ix_subscriptions_entity_type_entity_id; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX ix_subscriptions_entity_type_entity_id ON public.subscriptions USING btree (entity_type, entity_id); + + -- -- Name: ix_subscriptions_user_id; Type: INDEX; Schema: public; Owner: - -- @@ -14622,6 +14645,13 @@ CREATE INDEX mv_dashboard_transaction_types_idx ON public.mv_dashboard_transacti 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: - -- @@ -14832,6 +14862,13 @@ CREATE INDEX saves_item_idx ON public.saves USING btree (save_item_id, save_type CREATE INDEX saves_new_blocknumber_idx ON public.saves USING btree (blocknumber); +-- +-- Name: saves_user_created_at_active_idx; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX saves_user_created_at_active_idx ON public.saves USING btree (user_id, created_at DESC) INCLUDE (save_type, save_item_id) WHERE (is_delete = false); + + -- -- Name: saves_user_idx; Type: INDEX; Schema: public; Owner: - -- @@ -15228,7 +15265,7 @@ COMMENT ON INDEX public.sol_user_balances_mint_user_id_idx IS 'Index for quick a -- Name: subscriptions_current_uniq_idx; Type: INDEX; Schema: public; Owner: - -- -CREATE UNIQUE INDEX subscriptions_current_uniq_idx ON public.subscriptions USING btree (subscriber_id, user_id) WHERE (is_current = true); +CREATE UNIQUE INDEX subscriptions_current_uniq_idx ON public.subscriptions USING btree (subscriber_id, user_id, entity_type) WHERE (is_current = true); -- diff --git a/sql/03_migration_tracker.sql b/sql/03_migration_tracker.sql index 7dfc0bff..285d2995 100644 --- a/sql/03_migration_tracker.sql +++ b/sql/03_migration_tracker.sql @@ -114,7 +114,6 @@ functions/handle_comms_rpc_log.sql bca8170b77a97b521b050f49873d43d4 2026-05-27 0 functions/handle_dbc_pools.sql 5d8727fa203bb5f204868f4209a2022a 2026-05-27 00:22:36.618827+00 functions/handle_manager_request.sql 426004c1b9ac2be9e5721afb580679be 2026-05-27 00:22:36.848838+00 functions/handle_play.sql c710d1ef4805d7f99817baffc6ed8e25 2026-05-27 00:22:36.924069+00 -functions/handle_playlist.sql 4b338726d86db94fb3339e8407968cdf 2026-05-27 00:22:36.987976+00 functions/handle_playlist_track.sql bbb4dce9244617aa2d6580aae05d154a 2026-05-27 00:22:37.057245+00 functions/handle_reaction.sql 679796675e687b45288c517c568692d4 2026-05-27 00:22:37.13432+00 functions/handle_share.sql 84efa350bfd7f5501ccd34a93f6207a6 2026-05-27 00:22:37.371581+00 @@ -159,13 +158,13 @@ functions/handle_event.sql f15d9cc1838fa327b5df7e6b10c5ad7c 2026-07-28 05:42:28. functions/handle_follow.sql cf954862ef38daf93740ca1a88f24863 2026-07-28 05:42:28.792892+00 functions/handle_repost.sql e7cfd188b5aec2b01584dc8db1e9bc00 2026-07-28 05:42:28.962664+00 functions/handle_save.sql 05422848c57572704f78f27171e94058 2026-07-28 05:42:29.039008+00 -functions/handle_track.sql c2d4c5674b0cb1db907ad625fd957c91 2026-07-28 05:42:29.145055+00 functions/notify_on_row.sql a326d476636de01dd939047526b0cb92 2026-07-28 05:42:29.345724+00 preflight/0001_initial_block.sql 6cc3c0833c195a1104bed5bf849c0266 2026-07-28 05:42:29.588508+00 functions/handle_comment_reaction.sql 8153e3cdb922265857b6beaf20d29733 2026-05-30 01:37:22.856663+00 functions/handle_user_challenges.sql 202037a6ec14955885a479648e2cc57a 2026-08-05 00:50:26.86158+00 views/artist_coin_prices.sql fbfb4b530235c4b95f851bf5be2a063d 2026-08-05 00:50:27.089199+00 functions/handle_comment_thread.sql 6eb74eb92cf3a01421498df96c6832f3 2026-05-30 01:37:22.955997+00 +functions/handle_track.sql e7f09963e58d4462c8a43dbd4b11a275 2026-08-28 04:26:42.202584+00 functions/handle_eth_wallet_balance_change.sql 3e31160b4bc55e951d9dfa4d994c180b 2026-05-30 01:37:23.054573+00 functions/handle_fan_club_text_post.sql 531bf682bcfd67c6866faf8ccdf7603b 2026-05-30 01:37:23.160142+00 functions/handle_tastemaker.sql 04690b53bd094a59717ef3a5b5d2c0a0 2026-05-30 01:37:23.34542+00 @@ -215,6 +214,10 @@ migrations/0235_drop_coin_stats_shadow.sql ecbd5cd4d2edd02bbb86d760c9768388 2026 migrations/0236_saves_reposts_album_to_playlist.sql 5bd5036831dbfa656352dfd937c3fe5c 2026-08-05 00:50:26.077564+00 migrations/0237_users_one_current_row_backfill.sql b48ab562bc1ab92d12a17795a59cdf84 2026-08-05 00:50:26.167093+00 functions/handle_challenge_disbursements.sql 32db00f1ecfcfbda0094c0a5e1e6e300 2026-08-05 00:50:26.417396+00 +migrations/0238_backfill_track_playlist_reverse_index.sql 876450b97109942a25da304c101004fe 2026-08-28 04:26:41.602282+00 +migrations/0238_new_chain_queue_cursor.sql 2643e6250dd6b3ca4abb00fd8f417fca 2026-08-28 04:26:41.677782+00 +migrations/0239_saves_user_created_at_idx.sql 8a73b4eab0aa290ab720a706ec06e249 2026-08-28 04:26:41.748552+00 +functions/handle_playlist.sql 17abfed3041b8039dde8dfbfd025c5ff 2026-08-28 04:26:42.053363+00 \.