Wait for a held node without pinning a pool connection, and give up after a bounded wait naming it (hq ADR 0100)

internal/inventory/hold.go

@@ -2,8 +2,22 @@ package inventory
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"slices"
+	"strings"
 	"sync"
+	"time"
+)
+
+// ErrNodeBusy is a node another act kept holding for longer than a caller waits.
+var ErrNodeBusy = errors.New("another act is composing or sending that node's declaration")
+
+// HoldWaitFor is how long HoldNodes waits for a node another act holds, and HoldPoll how often it
+// looks again. Variables so a test need not wait minutes.
+var (
+	HoldWaitFor = 2 * time.Minute
+	HoldPoll    = 250 * time.Millisecond
 )
 
 // HoldNodes serialises composing and sending a declaration per node (novox/hq ADR 0100): while
@@ -11,16 +25,39 @@ import (
 // adopted could send that declaration after `converge --yes` sent the converged one, and the node
 // would return to adopted with nobody having asked.
 //
-// Session-level advisory locks on one connection, taken in name order so two callers holding
+// Session-level advisory locks on one connection, all or none: a set not wholly free is given back
-// overlapping sets cannot each wait on the other. Release gives every one back, and may be called
+// at once, so two callers holding overlapping sets never each wait on the other. **A waiter pins
-// more than once; a caller whose context ends while waiting holds nothing.
+// no connection.** It looks again every HoldPoll with a connection borrowed for the look, and gives
+// up after HoldWaitFor with ErrNodeBusy naming the node — so a stuck holder costs the pool one
+// connection, never one per caller queued behind it. Release gives every one back, and may be
+// called more than once.
 func (i *Inventory) HoldNodes(ctx context.Context, names []string) (func(), error) {
 	sorted := slices.Clone(names)
 	slices.Sort(sorted)
 	sorted = slices.Compact(sorted)
+	deadline := time.Now().Add(HoldWaitFor)
+	for {
+		release, busy, err := i.tryHold(ctx, sorted)
+		if err != nil || busy == "" {
+			return release, err
+		}
+		if time.Now().After(deadline) {
+			return nil, fmt.Errorf("%w: %s has been held for over %s — try again once it is done",
+				ErrNodeBusy, busy, HoldWaitFor)
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(HoldPoll):
+		}
+	}
+}
+
+// tryHold takes every named node's lock or none, and says which node was busy when it took none.
+func (i *Inventory) tryHold(ctx context.Context, sorted []string) (func(), string, error) {
 	conn, err := i.store.Pool().Acquire(ctx)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	var once sync.Once
 	release := func() {
@@ -36,12 +73,17 @@ func (i *Inventory) HoldNodes(ctx context.Context, names []string) (func(), erro
 		})
 	}
 	for _, name := range sorted {
-		if _, err := conn.Exec(ctx,
+		var took bool
-			`select pg_advisory_lock(hashtext('mesh-node-declaration:' || $1)::bigint)`,
+		if err := conn.QueryRow(ctx,
-			name); err != nil {
+			`select pg_try_advisory_lock(hashtext('mesh-node-declaration:' || $1)::bigint)`,
+			name).Scan(&took); err != nil {
 			release()
-			return nil, err
+			return nil, "", err
+		}
+		if !took {
+			release()
+			return nil, strings.TrimSpace(name), nil
 		}
 	}
-	return release, nil
+	return release, "", nil
 }

internal/inventory/hold_test.go

@@ -1,6 +1,8 @@
 package inventory
 
 import (
+	"errors"
+	"strings"
 	"testing"
 	"time"
 )
@@ -42,3 +44,45 @@ func TestHoldingANodeMakesTheNextHolderWait(t *testing.T) {
 		t.Fatal("releasing the node did not let the next holder in")
 	}
 }
+
+// A waiter pins no pool connection while it waits, and gives up after a bounded wait saying which
+// node is busy.
+func TestAWaiterPinsNoConnectionAndGivesUp(t *testing.T) {
+	inv := fresh(t)
+	ctx := t.Context()
+	release, err := inv.HoldNodes(ctx, []string{"anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer release()
+	savedWait, savedPoll := HoldWaitFor, HoldPoll
+	HoldWaitFor, HoldPoll = 1500*time.Millisecond, 50*time.Millisecond
+	defer func() { HoldWaitFor, HoldPoll = savedWait, savedPoll }()
+
+	pool := inv.store.Pool()
+	base := pool.Stat().AcquiredConns()
+	const waiters = 3
+	done := make(chan error, waiters)
+	for range waiters {
+		go func() {
+			_, err := inv.HoldNodes(ctx, []string{"anchor"})
+			done <- err
+		}()
+	}
+	// While they wait, the pool lends nothing to them for longer than a look.
+	pinned := 0
+	for range 10 {
+		time.Sleep(60 * time.Millisecond)
+		if n := int(pool.Stat().AcquiredConns() - base); n > pinned {
+			pinned = n
+		}
+	}
+	if pinned >= waiters {
+		t.Fatalf("%d connections were held by %d waiters", pinned, waiters)
+	}
+	for range waiters {
+		if err := <-done; !errors.Is(err, ErrNodeBusy) || !strings.Contains(err.Error(), "anchor") {
+			t.Fatalf("a waiter did not give up naming the busy node: %v", err)
+		}
+	}
+}