90 lines
3.0 KiB
Go
90 lines
3.0 KiB
Go
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
|
|
// one caller holds a node, another asking for it waits. Without it a push that composed a node as
|
|
// 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, 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
|
|
}
|
|
var once sync.Once
|
|
release := func() {
|
|
once.Do(func() {
|
|
// Unlocking all of this session's advisory locks, then handing the connection back: a
|
|
// connection returned still holding one would hold it for whoever borrows it next.
|
|
_, err := conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock_all()`)
|
|
if err != nil {
|
|
// The session's locks die with the session: close it rather than return it.
|
|
_ = conn.Conn().Close(context.WithoutCancel(ctx))
|
|
}
|
|
conn.Release()
|
|
})
|
|
}
|
|
for _, name := range sorted {
|
|
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
|
|
}
|
|
if !took {
|
|
release()
|
|
return nil, strings.TrimSpace(name), nil
|
|
}
|
|
}
|
|
return release, "", nil
|
|
}
|