fix(server): advisory lock released on the request context strands a task type in permanent already_running

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)
+	}
+}

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)
 	}
 }