terdut-server/internal/api/advisory_lock_test.go

package api

// This file is internal (package api, not api_test) because withAdvisoryLock is
// unexported and these tests exercise its locking semantics directly rather than
// through the full StartArchiver/StartNotifier loop, which would make the "does
// not run while held" case timing-dependent instead of deterministic. It opens a
// plain connection to TERDUT_TEST_DSN rather than reusing testdb_test.go's
// newTestDB, since that helper lives in the separate, already-compiled
// api_test package and a Postgres advisory lock needs no schema or migration
// to exercise.

import (
	"context"
	"database/sql"
	"os"
	"testing"

	_ "github.com/jackc/pgx/v5/stdlib"
)

// advisoryTestDB opens a plain, unmigrated connection to the test database. An
// unset DSN fails rather than skips, matching testdb_test.go's rationale: a
// suite that quietly tests nothing is worse than one that does not run.
func advisoryTestDB(t *testing.T) *sql.DB {
	t.Helper()

	dsn := os.Getenv("TERDUT_TEST_DSN")
	if dsn == "" {
		t.Fatalf("TERDUT_TEST_DSN is not set: these tests need Postgres.\n" +
			"Run `make test-db` for a local one, then\n" +
			"  export TERDUT_TEST_DSN=postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable")
	}

	db, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatalf("connect to TERDUT_TEST_DSN: %v", err)
	}
	t.Cleanup(func() { db.Close() })
	return db
}

func TestWithAdvisoryLock_RunsWhenFree(t *testing.T) {
	db := advisoryTestDB(t)
	ctx := context.Background()

	ran := false
	withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })

	if !ran {
		t.Fatal("fn did not run although the lock was free")
	}
}

func TestWithAdvisoryLock_SkipsWhileHeldElsewhere(t *testing.T) {
	db := advisoryTestDB(t)
	ctx := context.Background()

	// Hold the lock on a connection of our own, standing in for another
	// replica mid-pass.
	holder, err := db.Conn(ctx)
	if err != nil {
		t.Fatalf("acquire holder connection: %v", err)
	}
	defer holder.Close()
	if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_lock($1)", archiverLockKey); err != nil {
		t.Fatalf("pre-acquire lock: %v", err)
	}

	ran := false
	withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })
	if ran {
		t.Fatal("fn ran although another connection already held the lock")
	}

	if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", archiverLockKey); err != nil {
		t.Fatalf("release held lock: %v", err)
	}

	// Now that the holder released it, the next caller should get it.
	ran = false
	withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })
	if !ran {
		t.Fatal("fn did not run after the other connection released the lock")
	}
}

func TestWithAdvisoryLock_ReleasesAfterFnReturns(t *testing.T) {
	db := advisoryTestDB(t)
	ctx := context.Background()

	withAdvisoryLock(ctx, db, notifierLockKey, "test", func() {})

	// If the first call had leaked the lock, this one would see it held and
	// skip, leaving ran false.
	ran := false
	withAdvisoryLock(ctx, db, notifierLockKey, "test", func() { ran = true })
	if !ran {
		t.Fatal("fn did not run on a later call: the earlier call leaked its lock")
	}
}

func TestWithAdvisoryLock_KeysAreIndependent(t *testing.T) {
	db := advisoryTestDB(t)
	ctx := context.Background()

	holder, err := db.Conn(ctx)
	if err != nil {
		t.Fatalf("acquire holder connection: %v", err)
	}
	defer holder.Close()
	if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_lock($1)", archiverLockKey); err != nil {
		t.Fatalf("pre-acquire archiver lock: %v", err)
	}
	defer holder.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", archiverLockKey)

	// Holding archiverLockKey must not block notifierLockKey.
	ran := false
	withAdvisoryLock(ctx, db, notifierLockKey, "test", func() { ran = true })
	if !ran {
		t.Fatal("fn did not run under a different key although only archiverLockKey was held")
	}
}