musicbrainz-provider #2
@@ -131,17 +131,14 @@ func (a *App) Close() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (a *App) run(ctx context.Context) error {
|
func (a *App) run(ctx context.Context) error {
|
||||||
// Run an immediate sync+scan so the service produces results without
|
// Start the Web UI dashboard FIRST, in its own goroutine, so the dashboard
|
||||||
// waiting a full interval.
|
// accepts connections immediately. The initial sync below is throttled by
|
||||||
if err := a.doSync(ctx); err != nil {
|
// the MusicBrainz 1 req/s limit and can take many minutes on a large
|
||||||
if ctx.Err() != nil {
|
// library (worst case: a fresh DB where every artist needs MBID
|
||||||
return nil
|
// resolution) — exactly when an operator is most likely watching. Starting
|
||||||
}
|
// the server first means the dashboard is reachable (serving cached data)
|
||||||
log.Printf("Initial sync+scan failed: %v", err)
|
// during that window instead of refusing connections. It serves until ctx
|
||||||
}
|
// is cancelled, then shuts down gracefully.
|
||||||
|
|
||||||
// Start the Web UI dashboard in its own goroutine; it serves until ctx is
|
|
||||||
// cancelled, then shuts down gracefully.
|
|
||||||
if a.web != nil {
|
if a.web != nil {
|
||||||
go func() {
|
go func() {
|
||||||
if err := a.web.Start(ctx); err != nil {
|
if err := a.web.Start(ctx); err != nil {
|
||||||
@@ -157,7 +154,10 @@ func (a *App) run(ctx context.Context) error {
|
|||||||
// disabled (sender nil / enabled false), so always calling it is safe.
|
// disabled (sender nil / enabled false), so always calling it is safe.
|
||||||
a.startNotifier(ctx)
|
a.startNotifier(ctx)
|
||||||
|
|
||||||
// Kick off the periodic sync+scan loop goroutine.
|
// Kick off the periodic sync+scan loop. It runs an immediate first sync
|
||||||
|
// (governed by the same overlap guard as periodic ticks) so the service
|
||||||
|
// produces results without waiting a full interval, without racing a
|
||||||
|
// concurrent tick over the shared DB and rate-limited MusicBrainz client.
|
||||||
a.startPeriodicSync(ctx)
|
a.startPeriodicSync(ctx)
|
||||||
|
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
@@ -237,25 +237,38 @@ func (a *App) startPeriodicSync(ctx context.Context) {
|
|||||||
var free = make(chan struct{}, 1)
|
var free = make(chan struct{}, 1)
|
||||||
free <- struct{}{}
|
free <- struct{}{}
|
||||||
|
|
||||||
|
// launch starts a guarded sync if the slot is free, returning true when a
|
||||||
|
// sync was started and false when one is already in progress. The in-flight
|
||||||
|
// goroutine returns the token when done.
|
||||||
|
launch := func(label string) bool {
|
||||||
|
select {
|
||||||
|
case <-free:
|
||||||
|
go func() {
|
||||||
|
defer func() { free <- struct{}{} }()
|
||||||
|
if err := a.doSync(ctx); err != nil {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
log.Printf("%s sync+scan failed: %v", label, err)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Immediate first sync (guarded), so the service produces results without
|
||||||
|
// waiting a full interval and without racing the first ticker fire.
|
||||||
|
launch("Initial")
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
log.Println("Periodic sync stopped.")
|
log.Println("Periodic sync stopped.")
|
||||||
return
|
return
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
select {
|
if !launch("Periodic") {
|
||||||
case <-free:
|
|
||||||
// Slot was free; start a sync and release the slot when done.
|
|
||||||
go func() {
|
|
||||||
defer func() { free <- struct{}{} }()
|
|
||||||
if err := a.doSync(ctx); err != nil {
|
|
||||||
if ctx.Err() != nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
log.Printf("Periodic sync+scan failed: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
default:
|
|
||||||
// Previous sync still running; skip this tick.
|
// Previous sync still running; skip this tick.
|
||||||
log.Println("Skipping periodic sync: previous sync still in progress.")
|
log.Println("Skipping periodic sync: previous sync still in progress.")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -143,8 +143,10 @@ func TestStartPeriodicSync_CancelsCleanly(t *testing.T) {
|
|||||||
close(done)
|
close(done)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// With a 1h interval the ticker would never fire on its own; cancel should
|
// With a 1h interval the ticker never fires on its own; cancel should
|
||||||
// return promptly.
|
// return promptly. The loop does run one immediate (guarded) sync at
|
||||||
|
// startup, so depending on scheduling calls may be 0 (cancel won the race)
|
||||||
|
// or 1 (immediate sync ran) — but never more, since no tick can fire in 1h.
|
||||||
cancel()
|
cancel()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
@@ -154,8 +156,8 @@ func TestStartPeriodicSync_CancelsCleanly(t *testing.T) {
|
|||||||
t.Fatal("startPeriodicSync did not exit after ctx cancellation")
|
t.Fatal("startPeriodicSync did not exit after ctx cancellation")
|
||||||
}
|
}
|
||||||
|
|
||||||
if got := atomic.LoadInt64(&calls); got != 0 {
|
if got := atomic.LoadInt64(&calls); got > 1 {
|
||||||
t.Errorf("expected no sync calls with 1h interval, got %d", got)
|
t.Errorf("expected at most 1 (immediate) sync call with 1h interval, got %d", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -52,7 +52,7 @@ func GetExternalRelease(db *DB, rgid string) (*ExternalRelease, error) {
|
|||||||
var cachedAt sql.NullTime
|
var cachedAt sql.NullTime
|
||||||
var secondaryTypes sql.NullString
|
var secondaryTypes sql.NullString
|
||||||
err := db.Conn().QueryRow(
|
err := db.Conn().QueryRow(
|
||||||
"SELECT rgid, artist_id, title, type, release_date, is_ignored, cached_at, secondary_types FROM external_releases WHERE rgid = ?",
|
"SELECT rgid, artist_id, title, COALESCE(type,''), COALESCE(release_date,''), is_ignored, cached_at, secondary_types FROM external_releases WHERE rgid = ?",
|
||||||
rgid,
|
rgid,
|
||||||
).Scan(&r.RGID, &r.ArtistID, &r.Title, &r.Type, &r.ReleaseDate, &r.IsIgnored, &cachedAt, &secondaryTypes)
|
).Scan(&r.RGID, &r.ArtistID, &r.Title, &r.Type, &r.ReleaseDate, &r.IsIgnored, &cachedAt, &secondaryTypes)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -83,7 +83,7 @@ func SaveExternalRelease(db *DB, release *ExternalRelease) error {
|
|||||||
// GetExternalReleasesByArtist returns all external_release rows for a given artist_id.
|
// GetExternalReleasesByArtist returns all external_release rows for a given artist_id.
|
||||||
func GetExternalReleasesByArtist(db *DB, artistID string) ([]ExternalRelease, error) {
|
func GetExternalReleasesByArtist(db *DB, artistID string) ([]ExternalRelease, error) {
|
||||||
rows, err := db.Conn().Query(
|
rows, err := db.Conn().Query(
|
||||||
"SELECT rgid, artist_id, title, type, release_date, is_ignored, cached_at, secondary_types FROM external_releases WHERE artist_id = ?",
|
"SELECT rgid, artist_id, title, COALESCE(type,''), COALESCE(release_date,''), is_ignored, cached_at, secondary_types FROM external_releases WHERE artist_id = ?",
|
||||||
artistID,
|
artistID,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -116,7 +116,7 @@ func GetExternalReleasesByArtist(db *DB, artistID string) ([]ExternalRelease, er
|
|||||||
// GetIgnoredReleases returns all external_release rows where is_ignored = 1.
|
// GetIgnoredReleases returns all external_release rows where is_ignored = 1.
|
||||||
func GetIgnoredReleases(db *DB) ([]ExternalRelease, error) {
|
func GetIgnoredReleases(db *DB) ([]ExternalRelease, error) {
|
||||||
rows, err := db.Conn().Query(
|
rows, err := db.Conn().Query(
|
||||||
"SELECT rgid, artist_id, title, type, release_date, is_ignored, cached_at, secondary_types FROM external_releases WHERE is_ignored = 1",
|
"SELECT rgid, artist_id, title, COALESCE(type,''), COALESCE(release_date,''), is_ignored, cached_at, secondary_types FROM external_releases WHERE is_ignored = 1",
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("query ignored releases: %w", err)
|
return nil, fmt.Errorf("query ignored releases: %w", err)
|
||||||
|
|||||||
@@ -1,16 +1,37 @@
|
|||||||
package database
|
package database
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
|
||||||
|
sqlite3 "github.com/mattn/go-sqlite3"
|
||||||
)
|
)
|
||||||
|
|
||||||
// MarkNotificationSent records that a notification has been sent for the given RGID.
|
// MarkNotificationSent records that a notification has been sent for the given RGID.
|
||||||
|
//
|
||||||
|
// Uses INSERT OR IGNORE so a pre-existing marker for the same RGID (a
|
||||||
|
// same-second re-notify colliding on the (rgid, sent_at) primary key) is a
|
||||||
|
// no-op rather than an error: the marker's presence, not its exact timestamp,
|
||||||
|
// is what matters for idempotency.
|
||||||
|
//
|
||||||
|
// A concurrent re-sync that prunes the external_releases row before this insert
|
||||||
|
// would violate the FK constraint. OR IGNORE does NOT downgrade FK violations
|
||||||
|
// in this SQLite build, so the FK error is caught explicitly and treated as a
|
||||||
|
// benign no-op ("the release is already gone"). This ensures a single vanished
|
||||||
|
// release cannot abort a whole digest's mark-sent loop and trigger duplicate
|
||||||
|
// notifications on the next run.
|
||||||
func MarkNotificationSent(db *DB, rgid string) error {
|
func MarkNotificationSent(db *DB, rgid string) error {
|
||||||
_, err := db.Conn().Exec(
|
_, err := db.Conn().Exec(
|
||||||
"INSERT INTO notifications_sent (rgid) VALUES (?)",
|
"INSERT OR IGNORE INTO notifications_sent (rgid) VALUES (?)",
|
||||||
rgid,
|
rgid,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
var sqliteErr sqlite3.Error
|
||||||
|
if errors.As(err, &sqliteErr) && sqliteErr.Code == sqlite3.ErrConstraint &&
|
||||||
|
sqliteErr.ExtendedCode == sqlite3.ErrConstraintForeignKey {
|
||||||
|
// Release row was pruned concurrently; nothing to mark.
|
||||||
|
return nil
|
||||||
|
}
|
||||||
return fmt.Errorf("mark notification sent: %w", err)
|
return fmt.Errorf("mark notification sent: %w", err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -33,7 +54,7 @@ func IsNotificationSent(db *DB, rgid string) (bool, error) {
|
|||||||
// artists are excluded so the digest honors the monitoring contract.
|
// artists are excluded so the digest honors the monitoring contract.
|
||||||
func GetUnnotifiedReleases(db *DB) ([]ExternalRelease, error) {
|
func GetUnnotifiedReleases(db *DB) ([]ExternalRelease, error) {
|
||||||
rows, err := db.Conn().Query(`
|
rows, err := db.Conn().Query(`
|
||||||
SELECT e.rgid, e.artist_id, e.title, e.type, e.release_date, e.is_ignored
|
SELECT e.rgid, e.artist_id, e.title, COALESCE(e.type,''), COALESCE(e.release_date,''), e.is_ignored
|
||||||
FROM external_releases e
|
FROM external_releases e
|
||||||
JOIN artist_settings s ON e.artist_id = s.id
|
JOIN artist_settings s ON e.artist_id = s.id
|
||||||
LEFT JOIN notifications_sent n ON e.rgid = n.rgid
|
LEFT JOIN notifications_sent n ON e.rgid = n.rgid
|
||||||
|
|||||||
@@ -41,9 +41,38 @@ func TestMarkNotificationSent_New(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestMarkNotificationSent_DuplicateSecond verifies that inserting the same RGID twice
|
// TestMarkNotificationSent_MissingReleaseIsNoOp verifies that marking a release
|
||||||
// within the same second fails due to the composite primary key (rgid, sent_at).
|
// whose external_releases row does not exist (e.g. pruned by a concurrent
|
||||||
// In practice, notifications are sent at most once per day, so this is acceptable.
|
// re-sync) does not error: the FK violation is swallowed by INSERT OR IGNORE so
|
||||||
|
// a single vanished release cannot abort a digest's mark-sent loop.
|
||||||
|
func TestMarkNotificationSent_MissingReleaseIsNoOp(t *testing.T) {
|
||||||
|
db, err := New(":memory:")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("New() error: %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
// No artist/release inserted: rgid-gone has no external_releases row.
|
||||||
|
if err := MarkNotificationSent(db, "rgid-gone"); err != nil {
|
||||||
|
t.Fatalf("MarkNotificationSent() for missing release should be a no-op, got error: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Nothing should have been recorded (FK violation ignored, row skipped).
|
||||||
|
var count int
|
||||||
|
if err := db.Conn().QueryRow("SELECT COUNT(*) FROM notifications_sent WHERE rgid = ?", "rgid-gone").Scan(&count); err != nil {
|
||||||
|
t.Fatalf("count query error: %v", err)
|
||||||
|
}
|
||||||
|
if count != 0 {
|
||||||
|
t.Errorf("expected 0 notification rows for missing release, got %d", count)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestMarkNotificationSent_DuplicateSecond verifies that marking the same RGID
|
||||||
|
// twice within the same second is an idempotent no-op (INSERT OR IGNORE) rather
|
||||||
|
// than an error: a same-second collision on the composite primary key
|
||||||
|
// (rgid, sent_at) must not abort a digest's mark-sent loop, since that would
|
||||||
|
// leave later releases unmarked and cause duplicate notifications on the next
|
||||||
|
// run. The marker's presence, not its exact timestamp, is what matters.
|
||||||
func TestMarkNotificationSent_DuplicateSecond(t *testing.T) {
|
func TestMarkNotificationSent_DuplicateSecond(t *testing.T) {
|
||||||
db, err := New(":memory:")
|
db, err := New(":memory:")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -60,10 +89,9 @@ func TestMarkNotificationSent_DuplicateSecond(t *testing.T) {
|
|||||||
if err := MarkNotificationSent(db, "rgid-1"); err != nil {
|
if err := MarkNotificationSent(db, "rgid-1"); err != nil {
|
||||||
t.Fatalf("first MarkNotificationSent() error: %v", err)
|
t.Fatalf("first MarkNotificationSent() error: %v", err)
|
||||||
}
|
}
|
||||||
// Second insert in the same second should fail with a UNIQUE constraint error.
|
// Second mark in the same second should be a silent no-op, not an error.
|
||||||
err = MarkNotificationSent(db, "rgid-1")
|
if err := MarkNotificationSent(db, "rgid-1"); err != nil {
|
||||||
if err == nil {
|
t.Fatalf("duplicate MarkNotificationSent() should be a no-op, got error: %v", err)
|
||||||
t.Fatal("expected UNIQUE constraint error on duplicate insert, got nil")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Should still have exactly one row.
|
// Should still have exactly one row.
|
||||||
|
|||||||
@@ -93,9 +93,13 @@ func NotifyOnce(ctx context.Context, db *database.DB, sender Sender, cfg config.
|
|||||||
return 0, fmt.Errorf("notifier: send digest: %w", err)
|
return 0, fmt.Errorf("notifier: send digest: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The digest has already been delivered at this point. A failure to mark a
|
||||||
|
// single release must NOT abort the loop: doing so would leave later
|
||||||
|
// releases unmarked and cause them to be re-notified (duplicate digest) on
|
||||||
|
// the next run. Log and continue so every release in this batch is marked.
|
||||||
for _, m := range toNotify {
|
for _, m := range toNotify {
|
||||||
if err := database.MarkNotificationSent(db, m.RGID); err != nil {
|
if err := database.MarkNotificationSent(db, m.RGID); err != nil {
|
||||||
return 0, fmt.Errorf("notifier: mark sent for %s: %w", m.RGID, err)
|
log.Printf("notifier: mark sent for %s failed: %v", m.RGID, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return len(toNotify), nil
|
return len(toNotify), nil
|
||||||
|
|||||||
Reference in New Issue
Block a user