From aefe92c02e81a61aaf63cea2f94d5b1782996561 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Thu, 13 Aug 2026 12:26:05 -0700 Subject: [PATCH 1/2] fix(genesis-writer): emit release_date as RFC3339, not Postgres text The tracks query selected release_date::text, which yields Postgres's own "2026-09-06 22:06:00". The indexer's parseReleaseDate accepts RFC3339, RFC3339Nano and "Mon Jan 02 2006 15:04:05 GMT-0700" -- none of which match -- so releaseDateOrDefault silently fell back to block time, i.e. the source row's created_at. That is not a cosmetic date difference. A track whose release_date lands in the past is then picked up by ScheduledReleasePublisher, which sets is_unlisted = false and updated_at = now(). Measured on the 2026-08-07 snapshot: 372 unlisted tracks with a future release_date had the date rewritten 368 of them were published early 498 -> 136 tracks still holding a future release_date Verified end to end on track 2073330890: source release_date 2026-09-06 22:06:00 and is_unlisted true, the transaction carried both correctly, and the indexed row came out release_date 2025-08-10 20:57:26 (exactly its created_at) with is_unlisted false. Every other timestamp the writer emits already goes out as RFC3339; this one was the outlier. --- cmd/genesis-writer/entities_track.go | 21 ++++++++++-- .../release_date_format_test.go | 32 +++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) create mode 100644 cmd/genesis-writer/release_date_format_test.go diff --git a/cmd/genesis-writer/entities_track.go b/cmd/genesis-writer/entities_track.go index 2ab90525..4ea15927 100644 --- a/cmd/genesis-writer/entities_track.go +++ b/cmd/genesis-writer/entities_track.go @@ -94,7 +94,7 @@ type sourceTrack struct { IsUnlisted bool IsDownloadable bool IsOriginalAvailable bool - ReleaseDate *string + ReleaseDate *time.Time License *string ISRC *string ISWC *string @@ -157,7 +157,7 @@ func buildTrackMetadata(t sourceTrack, collaborators []int64) trackMetadataInner IsUnlisted: t.IsUnlisted, IsDownloadable: t.IsDownloadable, IsOriginalAvail: t.IsOriginalAvailable, - ReleaseDate: deref(t.ReleaseDate), + ReleaseDate: fmtReleaseDate(t.ReleaseDate), License: deref(t.License), ISRC: deref(t.ISRC), ISWC: deref(t.ISWC), @@ -211,6 +211,21 @@ func buildTrackMetadata(t sourceTrack, collaborators []int64) trackMetadataInner return inner } +// fmtReleaseDate emits RFC3339, the format the indexer's parseReleaseDate +// accepts. Selecting release_date::text instead yields Postgres's own +// "2026-09-06 22:06:00", which matches none of the accepted layouts, so the +// indexer silently fell back to block time. That is not a cosmetic date +// difference: a track whose release_date lands in the past is picked up by the +// scheduled-release publisher, which sets is_unlisted = false. On the +// 2026-08-07 snapshot 372 unlisted tracks with a future release date had the +// date rewritten and 368 of them were published early. +func fmtReleaseDate(t *time.Time) string { + if t == nil { + return "" + } + return t.UTC().Format(time.RFC3339) +} + func (w *Writer) writeTracks(ctx context.Context) error { // Pre-load collaborator lists so Track:Create metadata includes them, // which causes the ETL to create pending invites automatically. @@ -231,7 +246,7 @@ func (w *Writer) writeTracks(ctx context.Context) error { t.track_cid, t.cover_art, t.cover_art_sizes, t.preview_cid, t.is_unlisted, t.is_downloadable, t.is_original_available, - t.release_date::text, t.license, t.isrc, t.iswc, t.bpm, t.musical_key, + t.release_date, t.license, t.isrc, t.iswc, t.bpm, t.musical_key, t.is_custom_bpm, t.is_custom_musical_key, t.remix_of, t.stem_of, t.is_stream_gated, t.stream_conditions, diff --git a/cmd/genesis-writer/release_date_format_test.go b/cmd/genesis-writer/release_date_format_test.go new file mode 100644 index 00000000..1977fa10 --- /dev/null +++ b/cmd/genesis-writer/release_date_format_test.go @@ -0,0 +1,32 @@ +package main + +import ( + "testing" + "time" +) + +// The indexer's parseReleaseDate accepts RFC3339, RFC3339Nano and +// "Mon Jan 02 2006 15:04:05 GMT-0700" -- and nothing else. Selecting +// release_date::text yields Postgres's "2026-09-06 22:06:00", which matches +// none of them, so the indexer silently fell back to block time. A future +// release date rewritten into the past is then picked up by the +// scheduled-release publisher, which sets is_unlisted = false: on the +// 2026-08-07 snapshot that rewrote 372 dates and published 368 unlisted +// tracks early. +func TestReleaseDateIsEmittedInAnAcceptedLayout(t *testing.T) { + rd := time.Date(2026, 9, 6, 22, 6, 0, 0, time.UTC) + got := fmtReleaseDate(&rd) + + if _, err := time.Parse(time.RFC3339, got); err != nil { + t.Fatalf("fmtReleaseDate produced %q, which the indexer cannot parse: %v", got, err) + } + if got != "2026-09-06T22:06:00Z" { + t.Errorf("fmtReleaseDate = %q, want %q", got, "2026-09-06T22:06:00Z") + } + if got == "2026-09-06 22:06:00" { + t.Error("emitted Postgres text format, which no accepted layout matches") + } + if fmtReleaseDate(nil) != "" { + t.Errorf("nil release_date = %q, want empty so omitempty drops it", fmtReleaseDate(nil)) + } +} From f4c3e57446d2653609bbde91bda4a51bae70eec9 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:17:30 -0700 Subject: [PATCH 2/2] fix(genesis-writer): run events before comments Comments ran in phase 5 and events in phase 7, so every comment transaction preceded every event transaction. A comment carrying entity_type=Event is validated against the state its transaction lands on, and the indexer refuses it outright with "event %d does not exist". On the 2026-08-07 snapshot the source holds 69 Event comments. The writer emitted all of them -- they are on chain as transactions -- and the indexed database ended up with zero. Row counts on a 327k-row table hid it: 327,916 against 327,835. The dependency was already understood one line below, for event subscriptions ("the indexer rejects a subscription whose target event does not exist yet"). It was simply never applied to comments. Extracts the step table into Writer.steps() so the ordering is assertable, and adds TestStepOrderPutsReferencedEntitiesFirst covering the dependencies that are not obvious from reading the list. Restoring the old order fails it: step "events" runs after "comments" (positions 19, 16) Note this is distinct from the resume truncation that lost 36 events -- a clean re-run does not fix this one. --- cmd/genesis-writer/step_order_test.go | 46 ++++++++ cmd/genesis-writer/writer.go | 147 +++++++++++++++----------- 2 files changed, 134 insertions(+), 59 deletions(-) create mode 100644 cmd/genesis-writer/step_order_test.go diff --git a/cmd/genesis-writer/step_order_test.go b/cmd/genesis-writer/step_order_test.go new file mode 100644 index 00000000..85c49e81 --- /dev/null +++ b/cmd/genesis-writer/step_order_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "strings" + "testing" +) + +// The indexer validates references against the state a transaction lands on, +// so a step that emits rows pointing at entities a later step creates loses +// them outright -- silently, since row counts on the referencing table look +// plausible either way. +// +// Measured on the 2026-08-07 snapshot with comments running before events: +// all 69 Event comments were emitted by the writer and refused by the indexer +// with "event %d does not exist". +func TestStepOrderPutsReferencedEntitiesFirst(t *testing.T) { + order := (&Writer{cfg: &WriterConfig{}}).stepNames() + + idx := func(name string) int { + for i, n := range order { + if n == name { + return i + } + } + t.Fatalf("step %q not found in %v", name, order) + return -1 + } + + for _, dep := range []struct{ before, after, why string }{ + {"events", "comments", "a comment with entity_type=Event needs its event to exist"}, + {"events", "event subscriptions", "a subscription needs its target event to exist"}, + {"users", "tracks", "a track needs its owner"}, + {"tracks", "playlists", "playlist contents reference tracks"}, + {"comments", "comment reactions", "a reaction needs its comment"}, + {"comments", "comment pins", "a pin references a comment"}, + } { + if b, a := idx(dep.before), idx(dep.after); b > a { + t.Errorf("step %q runs after %q (positions %d, %d): %s", + dep.before, dep.after, b, a, dep.why) + } + } + + if strings.Join(order, ",") == "" { + t.Fatal("no steps registered") + } +} diff --git a/cmd/genesis-writer/writer.go b/cmd/genesis-writer/writer.go index a21dcb05..ed9fddbd 100644 --- a/cmd/genesis-writer/writer.go +++ b/cmd/genesis-writer/writer.go @@ -347,6 +347,93 @@ func (s *txCopySource) Err() error { return nil } // Run processes all entity types in dependency order, writes them as real // CometBFT blocks, then writes CometBFT state files. + +// writeStep is one phase of the run. Order matters: the indexer validates +// references against the state a transaction lands on, so a step emitting rows +// that point at entities a later step creates loses them silently. +type writeStep struct { + name string + skip bool + fn func(context.Context) error +} + +// steps is the ordered plan for a run. TestStepOrderPutsReferencedEntitiesFirst +// pins the dependencies that are not obvious from reading it. +func (w *Writer) steps() []writeStep { + return []writeStep{ + // Rewards first: it is the only step that depends on something outside + // this process (--core-dsn, pointed at the old chain, usually across a + // port-forward), and it is seconds of work against hours for the rest. + // Ordered last, an unreachable source failed the run only after all of + // that had completed. Nothing here depends on ordering — rewards + // reference only the pools this step creates, and no entity references + // a reward. + {"rewards", w.cfg.SkipRewards, w.writeRewards}, + + // Phase 1: Identity — users and their linked wallets + {"users", w.cfg.SkipUsers, w.writeUsers}, + {"associated wallets", w.cfg.SkipWallets, w.writeAssociatedWallets}, + {"dashboard wallet users", w.cfg.SkipWallets, w.writeDashboardWalletUsers}, + + // Phase 2: Content — tracks, collaborators, and playlists + {"tracks", w.cfg.SkipTracks, w.writeTracks}, + {"track collaborator approvals", w.cfg.SkipTracks, w.writeTrackCollaboratorApprovals}, + {"track downloads", w.cfg.SkipTracks, w.writeTrackDownloads}, + {"playlists", w.cfg.SkipPlaylists, w.writePlaylists}, + + // Phase 3: Social — relationships between users and content + {"follows", w.cfg.SkipSocial, w.writeFollows}, + {"saves", w.cfg.SkipSocial, w.writeSaves}, + {"reposts", w.cfg.SkipSocial, w.writeReposts}, + {"shares", w.cfg.SkipSocial, w.writeShares}, + {"subscriptions", w.cfg.SkipSocial, w.writeSubscriptions}, + {"muted users", w.cfg.SkipSocial, w.writeMutedUsers}, + + // Phase 4: Apps — developer apps and API grants + {"developer apps", w.cfg.SkipApps, w.writeDeveloperApps}, + {"grants", w.cfg.SkipApps, w.writeGrants}, + + // Phase 5: Events — before comments, which can hang off them. + // A comment carrying entity_type=Event is rejected outright if the + // event is not in place yet ("event %d does not exist"), and every + // comment transaction precedes every event transaction when this step + // runs later. On the 2026-08-07 snapshot that silently dropped all 69 + // Event comments -- the writer emitted them, the indexer refused them, + // and row counts on a 327k-row table hid it. + // Event subscriptions follow the events they point at for the same + // reason: the indexer rejects a subscription whose target event does + // not exist yet. + {"events", w.cfg.SkipEvents, w.writeEvents}, + {"event subscriptions", w.cfg.SkipEvents, w.writeEventSubscriptions}, + + // Phase 6: Comments — comments and reactions on content + {"comments", w.cfg.SkipComments, w.writeComments}, + {"comment reactions", w.cfg.SkipComments, w.writeCommentReactions}, + // A pin references both a comment and the track it is pinned to, so it + // has to follow the comments step and cannot run if either side was + // skipped. + {"comment pins", w.cfg.SkipComments || w.cfg.SkipTracks, w.writeCommentPins}, + + // Phase 7: Emails — encrypted emails and access grants + {"encrypted emails", w.cfg.SkipEmails, w.writeEncryptedEmails}, + {"email access", w.cfg.SkipEmails, w.writeEmailAccess}, + + // Phase 9: Activity — play count reconciliation and plays + {"play count reconciliation", w.cfg.SkipPlays, w.writePlayCountReconciliation}, + {"plays", w.cfg.SkipPlays, w.writePlays}, + } +} + +// stepNames returns the step order for tests. +func (w *Writer) stepNames() []string { + steps := w.steps() + names := make([]string, 0, len(steps)) + for _, s := range steps { + names = append(names, s.name) + } + return names +} + func (w *Writer) Run(ctx context.Context) error { start := time.Now() w.logger.Info("starting genesis write") @@ -436,65 +523,7 @@ func (w *Writer) Run(ctx context.Context) error { w.startBlockWriter(ctx) defer w.stopBlockWriter() //nolint:errcheck // explicit stop below captures the error - steps := []struct { - name string - skip bool - fn func(context.Context) error - }{ - // Rewards first: it is the only step that depends on something outside - // this process (--core-dsn, pointed at the old chain, usually across a - // port-forward), and it is seconds of work against hours for the rest. - // Ordered last, an unreachable source failed the run only after all of - // that had completed. Nothing here depends on ordering — rewards - // reference only the pools this step creates, and no entity references - // a reward. - {"rewards", w.cfg.SkipRewards, w.writeRewards}, - - // Phase 1: Identity — users and their linked wallets - {"users", w.cfg.SkipUsers, w.writeUsers}, - {"associated wallets", w.cfg.SkipWallets, w.writeAssociatedWallets}, - {"dashboard wallet users", w.cfg.SkipWallets, w.writeDashboardWalletUsers}, - - // Phase 2: Content — tracks, collaborators, and playlists - {"tracks", w.cfg.SkipTracks, w.writeTracks}, - {"track collaborator approvals", w.cfg.SkipTracks, w.writeTrackCollaboratorApprovals}, - {"track downloads", w.cfg.SkipTracks, w.writeTrackDownloads}, - {"playlists", w.cfg.SkipPlaylists, w.writePlaylists}, - - // Phase 3: Social — relationships between users and content - {"follows", w.cfg.SkipSocial, w.writeFollows}, - {"saves", w.cfg.SkipSocial, w.writeSaves}, - {"reposts", w.cfg.SkipSocial, w.writeReposts}, - {"shares", w.cfg.SkipSocial, w.writeShares}, - {"subscriptions", w.cfg.SkipSocial, w.writeSubscriptions}, - {"muted users", w.cfg.SkipSocial, w.writeMutedUsers}, - - // Phase 4: Apps — developer apps and API grants - {"developer apps", w.cfg.SkipApps, w.writeDeveloperApps}, - {"grants", w.cfg.SkipApps, w.writeGrants}, - - // Phase 5: Comments — comments and reactions on content - {"comments", w.cfg.SkipComments, w.writeComments}, - {"comment reactions", w.cfg.SkipComments, w.writeCommentReactions}, - // A pin references both a comment and the track it is pinned to, so it - // has to follow the comments step and cannot run if either side was - // skipped. - {"comment pins", w.cfg.SkipComments || w.cfg.SkipTracks, w.writeCommentPins}, - - // Phase 6: Emails — encrypted emails and access grants - {"encrypted emails", w.cfg.SkipEmails, w.writeEncryptedEmails}, - {"email access", w.cfg.SkipEmails, w.writeEmailAccess}, - - // Phase 7: Events - // Event subscriptions follow the events they point at: the indexer - // rejects a subscription whose target event does not exist yet. - {"events", w.cfg.SkipEvents, w.writeEvents}, - {"event subscriptions", w.cfg.SkipEvents, w.writeEventSubscriptions}, - - // Phase 9: Activity — play count reconciliation and plays - {"play count reconciliation", w.cfg.SkipPlays, w.writePlayCountReconciliation}, - {"plays", w.cfg.SkipPlays, w.writePlays}, - } + steps := w.steps() // Load completed steps for resume. completedSteps := make(map[string]bool)