diff --git a/internal/database/advisory_lock_integration_test.go b/internal/database/advisory_lock_integration_test.go new file mode 100644 index 000000000..a4aba3631 --- /dev/null +++ b/internal/database/advisory_lock_integration_test.go @@ -0,0 +1,78 @@ +//go:build integration +// +build integration + +package database_test + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestReleaseAdvisoryLock_DetachesFromCanceledCallerContext pins issue #105: +// ReleaseAdvisoryLock must actually release the session-level advisory lock +// even when the caller's context is already canceled or expired by the time +// the deferred release runs -- exactly what happens when a scheduled task's +// invocation context (Lambda deadline, or the HTTP request context) expires +// while dispatchTask is still running. +// +// Pre-fix, the unlock query ran on the caller's own ctx: a dead ctx made +// `SELECT pg_advisory_unlock($1)` fail immediately with "context canceled", +// which was only logged as a warning, and the otherwise-healthy pinned +// connection still went back to the pool via the deferred conn.Release() -- +// with the session-level lock still held on that backend session. Every +// subsequent TryAdvisoryLock for that task type would then see the lock +// held by that (still-alive, pooled) session and report "already_running" +// (HTTP 200) until that specific connection happened to be recycled, +// silently and indefinitely stalling the task. +// +// Verification uses a second, independent raw connection rather than +// container.DB's own pool: pg_try_advisory_lock is reentrant per session +// (a session re-acquiring a lock it already holds succeeds trivially), so +// asking container.DB.TryAdvisoryLock again could spuriously pass by +// reacquiring the very same pooled backend connection that never released +// the lock in the first place. +func TestReleaseAdvisoryLock_DetachesFromCanceledCallerContext(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequirePostgresContainer(ctx, t) + t.Cleanup(func() { _ = container.Cleanup(context.Background()) }) + + const lockID int64 = 424242424242 + + acquired, err := container.DB.TryAdvisoryLock(ctx, lockID) + require.NoError(t, err) + require.True(t, acquired, "must win the lock against a fresh session") + + // Simulate the caller's context already being dead by the time the + // deferred ReleaseAdvisoryLock call runs. + deadCtx, cancel := context.WithCancel(ctx) + cancel() + + container.DB.ReleaseAdvisoryLock(deadCtx, lockID) + + // Verify from an independent session: if the release actually ran + // (detached from deadCtx as it must), a different backend session can + // take the lock immediately. + verifyConn, err := pgx.Connect(ctx, container.Config.DSN(container.Config.Password)) + require.NoError(t, err) + defer func() { _ = verifyConn.Close(context.Background()) }() + + verifyCtx, verifyCancel := context.WithTimeout(ctx, 5*time.Second) + defer verifyCancel() + + var reacquired bool + err = verifyConn.QueryRow(verifyCtx, "SELECT pg_try_advisory_lock($1)", lockID).Scan(&reacquired) + require.NoError(t, err) + assert.True(t, reacquired, + "a different session must be able to take the lock: ReleaseAdvisoryLock must not skip "+ + "the unlock just because the caller's context is already canceled") + + if reacquired { + _, _ = verifyConn.Exec(verifyCtx, "SELECT pg_advisory_unlock($1)", lockID) + } +} diff --git a/internal/database/connection.go b/internal/database/connection.go index deb1c1ea6..c3485c04c 100644 --- a/internal/database/connection.go +++ b/internal/database/connection.go @@ -337,6 +337,11 @@ func (c *Connection) TryAdvisoryLock(ctx context.Context, lockID int64) (bool, e return true, nil } +// advisoryLockReleaseTimeout bounds the detached unlock query issued by +// ReleaseAdvisoryLock. Matches the 5s budget the scheduler's own deferred +// marker-clear uses for the same reason (scheduler.CollectRecommendations). +const advisoryLockReleaseTimeout = 5 * time.Second + // ReleaseAdvisoryLock releases a previously acquired advisory lock. // It reuses the same pinned connection used by TryAdvisoryLock to ensure // the unlock targets the correct PostgreSQL session. @@ -353,10 +358,37 @@ func (c *Connection) ReleaseAdvisoryLock(ctx context.Context, lockID int64) { } defer conn.Release() + // The caller's ctx is typically a deferred call sited right after + // TryAdvisoryLock (server/handler.go's HandleScheduledTask) and can + // already be canceled or past its deadline by the time this runs -- a + // long-running task whose Lambda invocation context (or HTTP request + // context) expired while dispatchTask was still in flight. Issuing the + // unlock on a dead context skips it entirely, and pg_advisory_unlock is + // session-scoped: the pinned connection is otherwise healthy, so it goes + // back to the pool with the lock still held, wedging every future + // acquire for this task type as "already_running" until that specific + // connection happens to be recycled (issue #105). Detach onto a fresh + // background context with a short timeout, matching the scheduler's own + // deferred marker-clear. + releaseCtx, cancel := context.WithTimeout(context.Background(), advisoryLockReleaseTimeout) + defer cancel() + var released bool - if err := conn.QueryRow(ctx, "SELECT pg_advisory_unlock($1)", lockID).Scan(&released); err != nil { + if err := conn.QueryRow(releaseCtx, "SELECT pg_advisory_unlock($1)", lockID).Scan(&released); err != nil { logging.Warnf("Failed to release advisory lock %d: %v", lockID, err) - } else if !released { + // The unlock query itself failed, so whether the session-level lock + // was actually dropped is unknown -- unlike the `!released` branch + // below, which means Postgres successfully confirmed no unlock was + // needed. A connection that might still hold the lock must never + // return to the pool: Conn.Release() reuses a connection unless it + // finds the underlying conn already closed, so close it here to + // force the deferred Release above to destroy it instead. + if closeErr := conn.Conn().Close(releaseCtx); closeErr != nil { + logging.Warnf("Failed to close advisory-lock connection %d after a failed unlock: %v", lockID, closeErr) + } + return + } + if !released { logging.Warnf("Advisory lock %d was not held by this session", lockID) } }