From 378155d09262c33a82c5c6652d1975107d3d258c Mon Sep 17 00:00:00 2001 From: Raymond Jacobson Date: Fri, 22 May 2026 09:59:19 -0700 Subject: [PATCH] fix(etl): playlist_seen PK, ErrNoRows on soft-deleted rows, attribute dispatch errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups from the overnight prod-clone validation run (38,834 chain blocks, 69 deep diff cycles against the read replica). 1. playlist_seen PK was wrong. Our 0005 migration declared `PRIMARY KEY (is_current, user_id, playlist_id, seen_at)` but prod's schema has `(user_id, playlist_id, seen_at)`. The handler's ON CONFLICT target that names the prod columns gets rejected with 42P10 against the old in-house PK; the version that names is_current too (which the previous PR had) silently mismatches prod. Fix both sides: - Update 0005 to declare the prod-correct PK on fresh databases. - Add migration 0027 to DROP+ADD the PK on databases that already ran the broken 0005. - Drop is_current from the ON CONFLICT spec in notification.go. 2. loadCurrentTrackRow / loadCurrentPlaylistRow propagated raw pgx.ErrNoRows when the target was soft-deleted (is_delete = true) between validation and update. The validation (trackExists / playlistExists) accepts is_delete=true rows; these row-loaders filter them out. Race manifested as bursts of "no rows in result set" at indexer.go:425 during high-Repost / high-Follow surge windows. Wrap ErrNoRows as ValidationError so the dispatcher logs at WARN, not ERROR. 3. getUserHandle in user_update.go had the same shape — propagated raw ErrNoRows in the rare race window between userExists and the handle lookup. Treat ErrNoRows as empty handle (callers compare against the new handle). 4. The dispatch-error log line at indexer.go:425 had no context fields, so when one fired we could see "no rows in result set" but had no clue which handler emitted it. Add entity_type / action / entity_id / user_id / hash to the log so future occurrences are self-attributing. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../migrations/0005_notification_tables.up.sql | 9 ++++++++- .../0027_fix_playlist_seen_pkey.down.sql | 3 +++ .../0027_fix_playlist_seen_pkey.up.sql | 17 +++++++++++++++++ pkg/etl/indexer.go | 13 ++++++++++++- .../processors/entity_manager/notification.go | 7 ++++++- .../processors/entity_manager/playlist_row.go | 7 +++++++ pkg/etl/processors/entity_manager/track_row.go | 9 +++++++++ .../processors/entity_manager/user_update.go | 9 +++++++++ 8 files changed, 71 insertions(+), 3 deletions(-) create mode 100644 pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql create mode 100644 pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql diff --git a/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql b/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql index 01cbd02f..ce55cf1a 100644 --- a/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql +++ b/pkg/etl/db/sql/migrations/0005_notification_tables.up.sql @@ -26,6 +26,13 @@ CREATE TABLE IF NOT EXISTS notification_seen ( CONSTRAINT notification_seen_pkey PRIMARY KEY (user_id, seen_at) ); +-- playlist_seen PK matches prod's schema: (user_id, playlist_id, seen_at). +-- The `is_current` column is retained for backward compatibility with the +-- legacy schema (always written as `true` by the playlist-seen handler), but +-- is NOT part of the uniqueness contract — including it in the PK would +-- mismatch prod and cause 42P10 on any ON CONFLICT clause that names the +-- prod target columns. Older databases that were created with the wrong PK +-- get fixed by migration 0027_fix_playlist_seen_pkey.up.sql. CREATE TABLE IF NOT EXISTS playlist_seen ( is_current boolean NOT NULL, user_id integer NOT NULL, @@ -34,5 +41,5 @@ CREATE TABLE IF NOT EXISTS playlist_seen ( blocknumber integer, blockhash character varying, txhash character varying, - CONSTRAINT playlist_seen_pkey PRIMARY KEY (is_current, user_id, playlist_id, seen_at) + CONSTRAINT playlist_seen_pkey PRIMARY KEY (user_id, playlist_id, seen_at) ); diff --git a/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql new file mode 100644 index 00000000..1bed2485 --- /dev/null +++ b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.down.sql @@ -0,0 +1,3 @@ +-- Reverting to the broken (is_current,...) PK is intentionally a no-op — +-- it would re-introduce the 42P10 incompatibility with prod and break the +-- ON CONFLICT in the playlist_seen handler. Leave the down side empty. diff --git a/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql new file mode 100644 index 00000000..ba4b6049 --- /dev/null +++ b/pkg/etl/db/sql/migrations/0027_fix_playlist_seen_pkey.up.sql @@ -0,0 +1,17 @@ +-- playlist_seen primary key was originally declared as +-- (is_current, user_id, playlist_id, seen_at) — `is_current` was a column +-- but never a meaningful part of the uniqueness contract (the handler +-- always writes is_current=true). Prod's schema has the PK without +-- is_current. ON CONFLICT clauses that name the prod target — which is +-- the right thing to do — get rejected with 42P10 against the older +-- in-house PK. +-- +-- Fresh databases skip this migration via the corrected 0005. This +-- migration converts older databases (that already ran the bad 0005) to +-- the prod-compatible shape. + +ALTER TABLE playlist_seen + DROP CONSTRAINT IF EXISTS playlist_seen_pkey; + +ALTER TABLE playlist_seen + ADD CONSTRAINT playlist_seen_pkey PRIMARY KEY (user_id, playlist_id, seen_at); diff --git a/pkg/etl/indexer.go b/pkg/etl/indexer.go index 312ae3bb..ef96e2b8 100644 --- a/pkg/etl/indexer.go +++ b/pkg/etl/indexer.go @@ -422,7 +422,18 @@ func (e *Indexer) indexBlocks() error { zap.String("hash", tx.Hash), ) } else { - e.logger.Error("entity manager dispatch error", zap.Error(dErr)) + // Include entity_type / action / hash so we can attribute + // non-validation handler errors. Without this context the + // log line is unreadable (e.g. a bare "no rows in result + // set" with no clue which handler emitted it). + e.logger.Error("entity manager dispatch error", + zap.String("entity_type", me.GetEntityType()), + zap.String("action", me.GetAction()), + zap.Int64("entity_id", me.GetEntityId()), + zap.Int64("user_id", me.GetUserId()), + zap.String("hash", tx.Hash), + zap.Error(dErr), + ) } } else { e.logger.Debug("tx indexed", diff --git a/pkg/etl/processors/entity_manager/notification.go b/pkg/etl/processors/entity_manager/notification.go index 41020a19..3e69d951 100644 --- a/pkg/etl/processors/entity_manager/notification.go +++ b/pkg/etl/processors/entity_manager/notification.go @@ -93,11 +93,16 @@ func (h *notificationViewPlaylistHandler) Handle(ctx context.Context, params *Pa return NewValidationError("playlist %d does not exist", params.EntityID) } + // PK on prod is (user_id, playlist_id, seen_at) — `is_current` is not + // part of the unique index, so it must not appear in the ON CONFLICT + // target (Postgres 42P10 otherwise). Observed firing during the + // prod-clone validation run: same user seeing the same playlist twice + // in the indexed window. _, err = params.DBTX.Exec(ctx, ` INSERT INTO playlist_seen ( is_current, user_id, playlist_id, seen_at, txhash, blocknumber ) VALUES (true, $1, $2, $3, $4, $5) - ON CONFLICT (is_current, user_id, playlist_id, seen_at) DO NOTHING + ON CONFLICT (user_id, playlist_id, seen_at) DO NOTHING `, params.UserID, params.EntityID, params.BlockTime, params.TxHash, params.BlockNumber) return err } diff --git a/pkg/etl/processors/entity_manager/playlist_row.go b/pkg/etl/processors/entity_manager/playlist_row.go index 68900ac4..264baa50 100644 --- a/pkg/etl/processors/entity_manager/playlist_row.go +++ b/pkg/etl/processors/entity_manager/playlist_row.go @@ -4,9 +4,11 @@ import ( "context" "database/sql" "encoding/json" + "errors" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) @@ -59,6 +61,11 @@ func loadCurrentPlaylistRow(ctx context.Context, dbtx db.DBTX, playlistID int64) &upc, &ddex, &ddexRel, &artists, ©right, &producerCopyright, &parentalWarning, &r.CreatedAt, ) + if errors.Is(err, pgx.ErrNoRows) { + // Same race as loadCurrentTrackRow: validation accepts is_delete=true + // rows but this loader filters them out. Convert to ValidationError. + return nil, NewValidationError("playlist %d is deleted or does not exist", playlistID) + } if err != nil { return nil, err } diff --git a/pkg/etl/processors/entity_manager/track_row.go b/pkg/etl/processors/entity_manager/track_row.go index cb169123..f00f55ea 100644 --- a/pkg/etl/processors/entity_manager/track_row.go +++ b/pkg/etl/processors/entity_manager/track_row.go @@ -4,9 +4,11 @@ import ( "context" "database/sql" "encoding/json" + "errors" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) @@ -68,6 +70,13 @@ func loadCurrentTrackRow(ctx context.Context, dbtx db.DBTX, trackID int64) (*tra &releaseDate, &r.IsScheduledRelease, &aiAttr, &r.IsPlaylistUpload, &ddex, &ddexRel, &r.IsAvailable, &segments, &r.CreatedAt, ) + if errors.Is(err, pgx.ErrNoRows) { + // Track was soft-deleted between validation and now (validateTrackUpdate + // uses trackExists which accepts is_delete=true; this function filters + // is_delete=false). Convert raw ErrNoRows to a ValidationError so the + // dispatcher logs it at WARN, not ERROR. + return nil, NewValidationError("track %d is deleted or does not exist", trackID) + } if err != nil { return nil, err } diff --git a/pkg/etl/processors/entity_manager/user_update.go b/pkg/etl/processors/entity_manager/user_update.go index d210e095..f0f71cfd 100644 --- a/pkg/etl/processors/entity_manager/user_update.go +++ b/pkg/etl/processors/entity_manager/user_update.go @@ -4,10 +4,12 @@ import ( "context" "database/sql" "encoding/json" + "errors" "strings" "time" "github.com/OpenAudio/go-openaudio/pkg/etl/db" + "github.com/jackc/pgx/v5" ) type userUpdateHandler struct{} @@ -258,6 +260,13 @@ func getCurrentUser(ctx context.Context, dbtx db.DBTX, userID int64) (*currentUs func getUserHandle(ctx context.Context, dbtx db.DBTX, userID int64) (string, error) { var handleLC sql.NullString err := dbtx.QueryRow(ctx, "SELECT handle_lc FROM users WHERE user_id = $1 AND is_current = true LIMIT 1", userID).Scan(&handleLC) + if errors.Is(err, pgx.ErrNoRows) { + // User has no current row (deleted between validation and now, or + // never existed under is_current=true). Treat as empty handle — + // callers compare against the new handle, so empty here means + // "different, run the uniqueness check". + return "", nil + } if handleLC.Valid { return handleLC.String, err }