diff --git a/internal/inventory/hold.go b/internal/inventory/hold.go index c9d2dd3..4ddf382 100644 --- a/internal/inventory/hold.go +++ b/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 -// overlapping sets cannot each wait on the other. Release gives every one back, and may be called -// more than once; a caller whose context ends while waiting holds nothing. +// Session-level advisory locks on one connection, all or none: a set not wholly free is given back +// at once, so two callers holding overlapping sets never each wait on the other. **A waiter pins +// 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, - `select pg_advisory_lock(hashtext('mesh-node-declaration:' || $1)::bigint)`, - name); err != nil { + var took bool + if err := conn.QueryRow(ctx, + `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 } diff --git a/internal/inventory/hold_test.go b/internal/inventory/hold_test.go index 70c110f..7de1d61 100644 --- a/internal/inventory/hold_test.go +++ b/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) + } + } +}