fix(explore): merge the catalog artifact in its own mbid encoding (#258) #259

Open
yonlu wants to merge 6 commits from fix/258-artifact-blob-cursor into main
3 changed files with 300 additions and 73 deletions
Showing only changes of commit d4ea14ca5c - Show all commits
+16
View File
@@ -845,6 +845,22 @@ that is quietly empty.
top-N, exact match, FTS search, popularity batch, the CAA map — and
asserts each returns something with a dashed id. A missed conversion
site shows up there and essentially nowhere else.
- **A comparison is typed on *both* sides, and a parameter is the half
that gets forgotten.** The paragraph above is about a literal; the
artifact merge positioned its batch walk with a Go `string` cursor
against the artifact's byte column, and SQLite answered rather than
complained: `mbid > ?` with a text key is true of every row, so the
bound the walk looked up was the same every time and the cursor
never advanced, while `mbid <= ?` is false of every row, so no batch
merged at all. The import looped indefinitely at 100% CPU behind a
progress bar reading "0 of 1,077,893 rows", merged nothing and
raised nothing (#258). Nothing caught it because the fixture that
guards the walk writes the old text form and the only compact one is
a single row — below `artifactMergeBatch`, so the bound query never
ran. `artifactKey` types the cursor to the artifact's own encoding
now, and the walk fails loudly when its bound does not strictly
advance, because the failure mode here is silence rather than a
wrong answer.
**The artifact is read in either encoding.** A published artifact
carries whichever form the exporter that built it used, and there is one
+69 -11
View File
@@ -1,8 +1,10 @@
package explore
import (
"bytes"
"context"
"database/sql"
"database/sql/driver"
"errors"
"fmt"
"os"
@@ -348,8 +350,12 @@ func (si *SearchIndex) analyzeIndex() {
// is an index range scan and a cancelled import leaves committed work
// behind rather than rolling it all back.
func (si *SearchIndex) mergeArtifactRows(ctx context.Context, total int) (int, error) {
// Asked once, because it is a property of the file and it decides
// how the walk's own comparisons are typed. See artifactKey.
storesText := si.artifactStoresText()
selectColumns := artifactSelectColumns(
si.artifactStoresText(), si.artifactHasTotals(),
storesText, si.artifactHasTotals(),
)
insertSQL := `
@@ -367,7 +373,7 @@ func (si *SearchIndex) mergeArtifactRows(ctx context.Context, total int) (int, e
WHERE mbid > ? AND mbid <= ?` + upsertIndexConflictSQL
var (
cursor string
cursor artifactKey
merged int
)
@@ -376,17 +382,30 @@ func (si *SearchIndex) mergeArtifactRows(ctx context.Context, total int) (int, e
return merged, err
}
upper, hasUpper, err := si.artifactBatchBound(cursor)
upper, hasUpper, err := si.artifactBatchBound(storesText, cursor)
if err != nil {
return merged, err
}
if hasUpper && bytes.Compare(upper, cursor) <= 0 {
// The predicate matched the cursor itself, so the walk can
// never advance. SQLite says nothing when a comparison is
// made between types it will not coerce - the query simply
// answers wrongly - so a mismatch here would otherwise spin
// forever behind an unmoving progress bar. Fail instead.
return merged, fmt.Errorf(
"%w: artifact walk did not advance past %x",
ErrArtifactUnusable, []byte(cursor),
)
}
var res sql.Result
if hasUpper {
res, err = si.db.ExecContext(insertRangeSQL, cursor, upper)
res, err = si.db.ExecContext(insertRangeSQL,
cursor.bind(storesText), upper.bind(storesText))
} else {
res, err = si.db.ExecContext(insertSQL, cursor)
res, err = si.db.ExecContext(insertSQL, cursor.bind(storesText))
}
if err != nil {
@@ -413,26 +432,65 @@ func (si *SearchIndex) mergeArtifactRows(ctx context.Context, total int) (int, e
}
}
// artifactKey is one MBID as the attached artifact stores it: 16 raw
// bytes in a compact artifact, the dashed 36-character form in one
// published before that storage change.
//
// It is a type with a bind method rather than a string because the
// comparison it feeds is typed, and the wrong type is silent. SQLite
// does not coerce between TEXT and BLOB and orders every blob after
// every text value, so a cursor bound as text against a byte column
// makes `mbid > ?` true of the whole table - the walk rediscovers the
// same batch bound forever, and `mbid <= ?` false of the whole table,
// so no batch merges at all. Nothing errors; the import simply never
// finishes. bind is the one place that knows which form the column is
// in, decided by artifactStoresText, which asks the artifact rather than
// trusting a version number.
type artifactKey []byte
// bind renders the key as a statement argument in the artifact's own
// encoding.
func (k artifactKey) bind(storesText bool) driver.Value {
if storesText {
return string(k)
}
// Never nil. database/sql converts a nil []byte to SQL NULL, and
// `mbid > NULL` is NULL for every row - so an unset cursor would
// agree with nothing and import nothing, which is the same silently
// empty merge this type exists to prevent, one type over.
if k == nil {
return []byte{}
}
return []byte(k)
}
// artifactBatchBound returns the MBID that ends the next batch, and
// whether one exists — no bound means the remainder is the last batch.
func (si *SearchIndex) artifactBatchBound(cursor string) (string, bool, error) {
var bound string
//
// The bound is read out of the artifact and handed back as an
// artifactKey, because it becomes the next comparison the walk makes.
func (si *SearchIndex) artifactBatchBound(
storesText bool, cursor artifactKey,
) (artifactKey, bool, error) {
var bound []byte
err := si.db.QueryRowWriter(
`SELECT mbid FROM core.explore_index
WHERE mbid > ? ORDER BY mbid LIMIT 1 OFFSET ?`,
cursor, artifactMergeBatch-1,
cursor.bind(storesText), artifactMergeBatch-1,
).Scan(&bound)
if errors.Is(err, sql.ErrNoRows) {
return "", false, nil
return nil, false, nil
}
if err != nil {
return "", false, fmt.Errorf("%w: batch bound: %w", ErrArtifactUnusable, err)
return nil, false, fmt.Errorf("%w: batch bound: %w", ErrArtifactUnusable, err)
}
return bound, true, nil
return artifactKey(bound), true, nil
}
// stampArtifactMeta records what the merge established: the catalog half
+215 -62
View File
@@ -1,6 +1,7 @@
package explore
import (
"bytes"
"context"
"database/sql"
"encoding/hex"
@@ -71,13 +72,7 @@ func writeTestArtifact(
}
}
for k, v := range meta {
if _, err := db.Exec(
`INSERT INTO artifact_meta (key, value) VALUES (?, ?)`, k, v,
); err != nil {
t.Fatalf("stamp artifact meta: %v", err)
}
}
stampArtifactMeta(t, db, meta)
for _, r := range rows {
if _, err := db.Exec(`
@@ -93,6 +88,101 @@ func writeTestArtifact(
return path
}
// compactArtifactSchema is the artifact cmd/indexexport publishes: the
// catalog's ids as 16 raw bytes, its entity types as codes, and the
// per-release-group total_tracks the exporter added after the first
// artifact was shipped.
//
// It matters that a fixture carries this encoding and not the older
// text one, because SQLite does not coerce between TEXT and BLOB and
// every comparison the importer makes against an mbid is therefore
// encoding-sensitive. writeTestArtifact above is the *other* fixture:
// it still writes the text form, which is what the first published
// artifact carries and what the importer must keep reading.
var compactArtifactSchema = []string{
`CREATE TABLE explore_index (
entity_type INTEGER NOT NULL,
mbid BLOB NOT NULL,
title TEXT NOT NULL,
artist_name TEXT NOT NULL,
artist_mbid BLOB NOT NULL,
aliases TEXT NOT NULL DEFAULT '',
popularity INTEGER NOT NULL DEFAULT 0,
listener_count INTEGER NOT NULL DEFAULT 0,
duration INTEGER NOT NULL DEFAULT 0,
caa_release_mbid BLOB NOT NULL DEFAULT x'',
release_name TEXT NOT NULL DEFAULT '',
primary_type TEXT NOT NULL DEFAULT '',
secondary_types TEXT NOT NULL DEFAULT '',
release_date TEXT NOT NULL DEFAULT '',
total_tracks INTEGER NOT NULL DEFAULT 0,
artist_type TEXT NOT NULL DEFAULT '',
country TEXT NOT NULL DEFAULT '',
disambiguation TEXT NOT NULL DEFAULT '',
sort_name TEXT NOT NULL DEFAULT '',
discog_fetched INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (mbid)
) WITHOUT ROWID`,
`CREATE TABLE artifact_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)`,
}
// writeCompactTestArtifact builds the artifact the exporter publishes
// today, in its own encoding, so the importer is exercised against what
// a client actually downloads rather than against what it was written
// for.
func writeCompactTestArtifact(
t *testing.T, meta map[string]string, rows []artifactRow,
) string {
t.Helper()
path := filepath.Join(t.TempDir(), "core-index.db")
db, err := sql.Open("sqlite", "file:"+path)
if err != nil {
t.Fatalf("open artifact: %v", err)
}
defer func() { _ = db.Close() }()
for _, stmt := range compactArtifactSchema {
if _, err := db.Exec(stmt); err != nil {
t.Fatalf("create artifact schema: %v", err)
}
}
stampArtifactMeta(t, db, meta)
for _, r := range rows {
if _, err := db.Exec(`
INSERT INTO explore_index
(entity_type, mbid, title, artist_name, artist_mbid, popularity)
VALUES (?, ?, ?, ?, ?, ?)`,
entityCode(r.entityType), mbidBytes(r.mbid), r.title,
r.artistName, mbidBytes(r.artistMBID), r.popularity,
); err != nil {
t.Fatalf("insert artifact row: %v", err)
}
}
return path
}
// stampArtifactMeta writes the artifact_meta rows a fixture declares.
func stampArtifactMeta(t *testing.T, db *sql.DB, meta map[string]string) {
t.Helper()
for k, v := range meta {
if _, err := db.Exec(
`INSERT INTO artifact_meta (key, value) VALUES (?, ?)`, k, v,
); err != nil {
t.Fatalf("stamp artifact meta: %v", err)
}
}
}
// validMeta is the artifact_meta a well-formed artifact carries.
func validMeta() map[string]string {
return map[string]string{
@@ -270,6 +360,71 @@ func TestImportCoreArtifactBatchWalkCoversAllRows(t *testing.T) {
}
}
// TestImportCoreArtifactBatchWalkCoversAllRowsCompact is the batch walk
// on the encoding the exporter actually publishes.
//
// The walk positions itself by comparing the artifact's own mbid column
// against the last id it reached, and that column holds 16 raw bytes.
// SQLite does not coerce between TEXT and BLOB, and a blob sorts after
// every text value, so a cursor bound as text is a predicate that either
// matches every row or none: `mbid > ?` with an empty text key is true
// of the whole table, so
// the 100th row is always the 100th row and the bound never advances,
// while `mbid <= <text>` is false of the whole table, so no batch ever
// merges. The result is not a wrong import but an unbounded loop that
// merges nothing and never fails.
//
// Both encodings are covered on purpose. The walk was only ever tested
// against the text fixture above, which is why it shipped broken on the
// one the clients download.
func TestImportCoreArtifactBatchWalkCoversAllRowsCompact(t *testing.T) {
db := database.NewTestDB(t)
si := NewSearchIndex(db, nil, nil, testLogger())
original := artifactMergeBatch
artifactMergeBatch = 100
t.Cleanup(func() { artifactMergeBatch = original })
const total = 337
rows := make([]artifactRow, 0, total)
for i := range total {
rows = append(rows, artifactRow{
entityType: EntityRecording,
mbid: syntheticMBID(i),
title: "Song",
artistName: "Artist",
artistMBID: artA,
popularity: i,
})
}
path := writeCompactTestArtifact(t, validMeta(), rows)
if err := si.importCoreArtifact(context.Background(), path); err != nil {
t.Fatalf("importCoreArtifact: %v", err)
}
var got, top int
if err := db.QueryRowWriter(
"SELECT COUNT(*), MAX(popularity) FROM explore_index",
).Scan(&got, &top); err != nil {
t.Fatalf("count rows: %v", err)
}
if got != total {
t.Errorf("merged %d rows, want %d", got, total)
}
// A count alone would pass if the walk re-merged the same first
// batch forever, so the far end of the artifact is checked too.
if top != total-1 {
t.Errorf("highest popularity = %d, want %d", top, total-1)
}
}
func TestImportCoreArtifactRejectsBadArtifacts(t *testing.T) {
tests := []struct {
name string
@@ -454,61 +609,9 @@ func TestArtifactColumnsMatchExporter(t *testing.T) {
// the importer decides by asking the artifact, not by trusting a
// version number, and both must land identically.
func TestImportCoreArtifactAcceptsBothEncodings(t *testing.T) {
compact := filepath.Join(t.TempDir(), "core-index.db")
db, err := sql.Open("sqlite", "file:"+compact)
if err != nil {
t.Fatalf("open artifact: %v", err)
}
if _, err := db.Exec(`CREATE TABLE explore_index (
entity_type INTEGER NOT NULL,
mbid BLOB NOT NULL,
title TEXT NOT NULL,
artist_name TEXT NOT NULL,
artist_mbid BLOB NOT NULL,
aliases TEXT NOT NULL DEFAULT '',
popularity INTEGER NOT NULL DEFAULT 0,
listener_count INTEGER NOT NULL DEFAULT 0,
duration INTEGER NOT NULL DEFAULT 0,
caa_release_mbid BLOB NOT NULL DEFAULT x'',
release_name TEXT NOT NULL DEFAULT '',
primary_type TEXT NOT NULL DEFAULT '',
secondary_types TEXT NOT NULL DEFAULT '',
release_date TEXT NOT NULL DEFAULT '',
artist_type TEXT NOT NULL DEFAULT '',
country TEXT NOT NULL DEFAULT '',
disambiguation TEXT NOT NULL DEFAULT '',
sort_name TEXT NOT NULL DEFAULT '',
discog_fetched INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (mbid)
)`); err != nil {
t.Fatalf("create artifact table: %v", err)
}
if _, err := db.Exec(
`CREATE TABLE artifact_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)`,
); err != nil {
t.Fatalf("create artifact meta: %v", err)
}
for k, v := range validMeta() {
if _, err := db.Exec(
"INSERT INTO artifact_meta (key, value) VALUES (?, ?)", k, v,
); err != nil {
t.Fatalf("write artifact meta: %v", err)
}
}
if _, err := db.Exec(`
INSERT INTO explore_index (entity_type, mbid, title, artist_name, artist_mbid, popularity)
VALUES (1, ?, 'Artist A', 'Artist A', ?, 5000)`,
mbidBytes(artA), mbidBytes(artA),
); err != nil {
t.Fatalf("write artifact row: %v", err)
}
_ = db.Close()
compact := writeCompactTestArtifact(t, validMeta(), []artifactRow{
{EntityArtist, artA, "Artist A", "Artist A", artA, 5000},
})
live := database.NewTestDB(t)
si := NewSearchIndex(live, nil, nil, testLogger())
@@ -748,3 +851,53 @@ func TestImportCoreArtifactWithoutCredits(t *testing.T) {
t.Errorf("credit refs = %d, want 0", refs)
}
}
// TestArtifactKeyBindsInTheArtifactsOwnEncoding pins the one place the
// batch walk's comparison type is decided.
//
// Every wrong answer is silent, which is why it is worth pinning all
// four. SQLite does not coerce TEXT to BLOB and orders every blob after
// every text value, so a text key against a byte column makes
// `mbid > ?` true of the whole artifact - the cursor never advances and
// the walk spins forever without merging a row - while a byte key
// against a text column makes it false of the whole artifact, so every
// batch merges nothing and the import "succeeds" empty. An unset cursor
// is the same fault once more: database/sql converts a nil []byte to
// SQL NULL, and `mbid > NULL` matches no row at all.
func TestArtifactKeyBindsInTheArtifactsOwnEncoding(t *testing.T) {
raw := mbidBytes(artA)
for _, tt := range []struct {
name string
key artifactKey
want []byte
}{
{"unset", nil, []byte{}},
{"set", artifactKey(raw), raw},
} {
t.Run("bytes/"+tt.name, func(t *testing.T) {
got, ok := tt.key.bind(false).([]byte)
if !ok {
t.Fatalf("bind(false) = %T, want []byte", tt.key.bind(false))
}
if got == nil {
t.Fatal("bound to SQL NULL, which matches no row")
}
if !bytes.Equal(got, tt.want) {
t.Errorf("bind(false) = %x, want %x", got, tt.want)
}
})
}
// The dashed form is what an artifact published before the storage
// change carries, and it has to compare as text against text.
if got := artifactKey(nil).bind(true); got != "" {
t.Errorf("bind(true) on an unset cursor = %#v, want an empty string", got)
}
if got := artifactKey(artA).bind(true); got != artA {
t.Errorf("bind(true) = %#v, want %q", got, artA)
}
}