test(storage): force the copy/upload race so the storage lock's test can go red (TASK-3779) (#2135)

* test(storage): force the copy/upload race so the storage lock's test can go red (TASK-3779)

The 25-round race between a cross-workspace copy and an upload (BUG-3653)
passed with the storage advisory lock removed on Postgres: the long copy
and the short upload never overlapped inside the check-to-commit window.

testHookCopyAfterStorageCheck holds the copy after its storage check and
before its inserts, with the lock held. The upload starts in that window.
With the lock it waits for the copy to commit and is refused. Without it
(mutant on test-pg) it commits while the copy is held, and the test fails
naming that. Nil in production.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* test(storage): the forced race proves the upload reached the lock, and never strands a held copy (TASK-3779)

Codex round 1:
- The 500ms window could pass vacuously if the upload had not reached
  the lock yet. testHookAfterStorageOwnerRead is now armed once the copy
  is held, so it fires for the upload alone, just before it takes the
  lock, and the window starts there.
- A failure before release left the copy parked holding its locks.
  Cleanup now releases it once and joins the copy goroutine.

The test is Postgres-only: on SQLite the upload's BEGIN IMMEDIATE waits
behind the held copy's writer lock before reaching the advisory lock it
watches.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* test(storage): prove the upload waits on the advisory lock from pg_locks, and join it on cleanup (TASK-3779)

Codex round 2: the owner-read hook fires a few statements before the
lock wait, so it did not prove the upload was blocked on the lock. The
test now polls pg_locks for an ungranted advisory lock in its own
database while the copy is held: present with the lock, never present
without it (the upload then commits past the cap and the test fails).
Cleanup also joins the upload goroutine.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* test(storage): the waiter check matches the storage lock's own key (TASK-3779)

Codex round 3: any ungranted advisory lock in the database counted as
the upload's wait. The pg_locks query now matches objid to the low 32
bits of hashtext('pad:plan-limit:storage_bytes:' || owner), with
objsubid 1.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

* test(storage): match the storage lock's classid as well as its objid (TASK-3779)

Codex round 4: objid alone is the low 32 bits of the key.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

internal/store/attachments.go

@@ -2141,6 +2141,12 @@ var ErrStorageOwnerMoved = errStorageOwnerMoved
 // SQLite nothing else can write while it is open. Nil in production.
 var testHookAfterStorageOwnerRead func(tx *sql.Tx, workspaceID string)
 
+// testHookCopyAfterStorageCheck, when set by a test, runs in a cross-workspace
+// copy after its storage check passed and before it inserts anything, with
+// the storage lock held: the window a concurrent upload must wait out
+// (TASK-3779). Nil in production.
+var testHookCopyAfterStorageCheck func()
+
 // enforceStorageLimitTx is the authoritative storage check (TASK-3619): it
 // refuses an insert of adding bytes that would take the workspace past its
 // owner's storage_bytes limit. Usage is what WorkspaceStorageInfo sums: the

internal/store/bug3653_copy_storage_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/PerpetualSoftware/pad/internal/models"
 )
@@ -181,3 +182,95 @@ func TestBUG3653_CopyRetriesAnOwnershipChangeMidCheck(t *testing.T) {
 		t.Fatalf("attempts = %d, want 3", calls)
 	}
 }
+
+// TASK-3779: the race above never overlapped the two writes, so it passed
+// with the storage lock removed. Here the copy is held after its check passes
+// and before it inserts, with room for only one of the two 600-byte writes,
+// and the upload starts inside that window. With the lock the upload waits for
+// the copy to commit and is refused; without it both commit past the cap.
+func TestTASK3779_UploadWaitsForAHeldCopyAndIsRefused(t *testing.T) {
+	f := newCopyStorageFixture(t, "pro", 600, 0)
+	if f.s.dialect.Driver() != DriverPostgres {
+		// The advisory lock under test is Postgres-only. On SQLite the
+		// upload's BEGIN IMMEDIATE waits behind the held copy's writer lock
+		// before it can reach the lock this test watches.
+		t.Skip("the storage advisory lock is Postgres-only")
+	}
+	used, err := f.s.accountStorageUsageQ(f.s.db, f.owner.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := f.s.SetUserPlanOverrides(f.owner.ID, fmt.Sprintf(`{"storage_bytes":%d}`, used+1000)); err != nil {
+		t.Fatal(err)
+	}
+	held, release := make(chan struct{}), make(chan struct{})
+	var releaseOnce sync.Once
+	releaseCopy := func() { releaseOnce.Do(func() { close(release) }) }
+	testHookCopyAfterStorageCheck = func() {
+		close(held)
+		<-release
+	}
+	copyErr, uploadErr := make(chan error, 1), make(chan error, 1)
+	copyDone, uploadDone := make(chan struct{}), make(chan struct{})
+	uploadStarted := false
+	// Failure paths must not leave the copy parked holding its locks, or a
+	// goroutine running past the test.
+	t.Cleanup(func() {
+		releaseCopy()
+		<-copyDone
+		if uploadStarted {
+			<-uploadDone
+		}
+		testHookCopyAfterStorageCheck = nil
+	})
+
+	go func() { defer close(copyDone); copyErr <- f.copy(true) }()
+	select {
+	case <-held:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the copy never reached its storage check")
+	}
+	uploadStarted = true
+	go func() { defer close(uploadDone); uploadErr <- f.add(600, false, WithPlanLimit()) }()
+
+	// The upload must be seen WAITING on the owner's storage lock, in
+	// Postgres's own lock table, while the copy is held: the proof that the
+	// lock serialises them, not merely that the upload was slow. Without the
+	// lock no waiter ever appears and the upload commits past the cap.
+	waiting := false
+	deadline := time.Now().Add(10 * time.Second)
+	for !waiting && time.Now().Before(deadline) {
+		select {
+		case err := <-uploadErr:
+			t.Fatalf("the upload finished while the copy was held between its check and its commit (err %v): the storage lock does not serialise them", err)
+		case <-time.After(20 * time.Millisecond):
+		}
+		var n int
+		// The storage lock's own key: acquirePlanLimitLock takes
+		// pg_advisory_xact_lock(hashtext('pad:plan-limit:storage_bytes:' ||
+		// owner)), a single bigint key that pg_locks splits into classid (high
+		// 32 bits) and objid (low 32 bits), with objsubid 1. Both are matched.
+		if err := f.s.db.QueryRow(`SELECT COUNT(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted
+			AND database = (SELECT oid FROM pg_database WHERE datname = current_database())
+			AND objsubid = 1
+			AND classid = (((hashtext('pad:plan-limit:storage_bytes:' || $1)::bigint) >> 32) & 4294967295)::oid
+			AND objid = ((hashtext('pad:plan-limit:storage_bytes:' || $1)::bigint) & 4294967295)::oid`, f.owner.ID).Scan(&n); err != nil {
+			t.Fatal(err)
+		}
+		waiting = n > 0
+	}
+	if !waiting {
+		t.Fatal("the upload never waited on the storage lock while the copy was held")
+	}
+	releaseCopy()
+	if err := <-copyErr; err != nil {
+		t.Fatalf("the held copy: %v", err)
+	}
+	var ple *PlanLimitError
+	if err := <-uploadErr; !errors.As(err, &ple) {
+		t.Fatalf("the upload after the copy committed: err = %v, want *PlanLimitError", err)
+	}
+	if usage, _ := f.s.accountStorageUsageQ(f.s.db, f.owner.ID); usage > used+1000 {
+		t.Fatalf("usage %d is past the limit %d", usage, used+1000)
+	}
+}

internal/store/items_cross_workspace_copy.go

@@ -790,6 +790,9 @@ func (s *Store) copyItemAcrossWorkspacesTx(req CrossWorkspaceCopyRequest, source
 		if err := s.checkStorageLimitTx(tx, storageLock, adding); err != nil {
 			return nil, err
 		}
+		if testHookCopyAfterStorageCheck != nil {
+			testHookCopyAfterStorageCheck()
+		}
 	}
 	if len(plan.UnresolvableRefs) > 0 {
 		// DR-11a observability: a spike means either a data-integrity problem