Documentation
¶
Overview ¶
Package mutex implements the on-disk lease store for agentmutex.
Layout under the state root (default ~/.agentmutex):
locks/<encoded-key>/guard per-key mutation guard (flock target) locks/<encoded-key>/holder.json the current lease, if any locks/<encoded-key>/queue/ FIFO waiter entries
Every mutation (acquire, release, renew, prune) runs while holding the per-key guard, so read-modify-write sequences are serialized across processes without any daemon. holder.json is replaced via temp-file + rename, so guard-free readers (status, list) always see a complete document.
Waiter entries are named "<zero-padded-unixnano>-<token>.json", so lexicographic order is arrival order. A waiter proves it is alive by touching its entry's mtime on every poll; entries whose mtime is older than WaiterStaleAfter belong to dead waiters and are skipped.
Index ¶
- Constants
- Variables
- func DecodeKey(name string) (string, error)
- func EncodeKey(key string) string
- func NewToken() string
- func ValidateKey(key string) error
- type AcquireOpts
- type AcquireResult
- type CorruptError
- type Holder
- type KeyStatus
- type NotHolderError
- type PruneResult
- type Store
- func (s *Store) Dequeue(entryPath string) error
- func (s *Store) Enqueue(key string, w Waiter) (string, error)
- func (s *Store) Heartbeat(entryPath string, w Waiter) error
- func (s *Store) List() ([]KeyStatus, error)
- func (s *Store) Prune() (*PruneResult, error)
- func (s *Store) Release(key, token string, force bool) (*Holder, error)
- func (s *Store) Renew(key, token string, ttl time.Duration) (*Holder, error)
- func (s *Store) Status(key string) (*KeyStatus, error)
- func (s *Store) TryAcquire(key, token string, o AcquireOpts) (*AcquireResult, error)
- type Waiter
Constants ¶
const ( DefaultTTL = 15 * time.Minute DefaultPollInterval = 1 * time.Second DefaultWaiterStaleAfter = 30 * time.Second )
Defaults. Deploy-sized critical sections run minutes, and waiting agents are inference-bound (they think for seconds to minutes between actions), so generous TTLs and second-scale polling are free.
const MaxKeyLen = 80
MaxKeyLen bounds key length so encoded directory names stay well under filesystem name limits (255 bytes) even when every byte is %XX-escaped.
Variables ¶
var ErrNotHeld = errors.New("no lease held for this key")
ErrNotHeld is returned when releasing or renewing a key that has no lease.
Functions ¶
func EncodeKey ¶
EncodeKey converts a semantic key into a filesystem-safe directory name. Bytes outside [A-Za-z0-9._-] are percent-encoded, so the mapping is reversible and safe on every platform (":" is not legal on Windows). Windows-reserved device names and trailing dots are escaped too.
func NewToken ¶
func NewToken() string
NewToken returns a fresh lease token: 128 bits of randomness, hex-encoded. Only the holder of the token can release or renew the lease.
func ValidateKey ¶
ValidateKey checks that key is a usable semantic key. Keys are structured namespaces such as "deploy:staging" or "account:12345:balance". They must start with an alphanumeric character and may contain letters, digits and the separators ". _ - : /".
Types ¶
type AcquireOpts ¶
type AcquireOpts struct {
TTL time.Duration
Agent string
PID int
Host string
Reason string
// Enqueued indicates the token has a queue entry that should be
// removed if the acquire succeeds.
Enqueued bool
}
AcquireOpts parameterizes TryAcquire.
type AcquireResult ¶
type AcquireResult struct {
Acquired bool
// Holder is our new lease when Acquired, otherwise the current
// unexpired holder blocking us (nil if the key is free).
Holder *Holder
// Blocker is the fresh queue head ahead of us, when the key is free
// (or expired) but FIFO order says it is not our turn.
Blocker *Waiter
}
AcquireResult reports one TryAcquire attempt.
type CorruptError ¶
CorruptError is returned when a state file exists but cannot be parsed. The store never steals a lease it cannot read; use force-release to clear.
func (*CorruptError) Error ¶
func (e *CorruptError) Error() string
type Holder ¶
type Holder struct {
Key string `json:"key"`
Token string `json:"token,omitempty"`
Agent string `json:"agent"`
PID int `json:"pid"`
Host string `json:"host"`
Reason string `json:"reason,omitempty"`
AcquiredAt time.Time `json:"acquired_at"`
ExpiresAt time.Time `json:"expires_at"`
RenewedAt *time.Time `json:"renewed_at,omitempty"`
}
Holder is a lease on a semantic key. Token is omitempty so display paths can redact it (the CLI blanks tokens in status/list output).
type KeyStatus ¶
type KeyStatus struct {
Key string `json:"key"`
State string `json:"state"` // "held", "expired" or "free"
Holder *Holder `json:"holder,omitempty"`
Waiters []Waiter `json:"waiters"`
}
KeyStatus is a read-only snapshot of one key.
type NotHolderError ¶
type NotHolderError struct{ Holder *Holder }
NotHolderError is returned when the presented token does not match the current lease. It carries the actual holder for diagnostics.
func (*NotHolderError) Error ¶
func (e *NotHolderError) Error() string
type PruneResult ¶
type PruneResult struct {
ExpiredLeases []string `json:"expired_leases"`
StaleWaiters int `json:"stale_waiters"`
}
PruneResult reports what Prune removed.
type Store ¶
type Store struct {
Root string
WaiterStaleAfter time.Duration
// contains filtered or unexported fields
}
Store is an on-disk lease store. Safe for concurrent use by any number of processes on the same machine. (Not across hosts: the guard is flock(2) and waiter liveness compares this host's clock against entry mtimes.)
func Open ¶
Open opens (creating if needed) the store at root. An empty root falls back to $AGENTMUTEX_DIR, then ~/.agentmutex.
func (*Store) Dequeue ¶
Dequeue removes a waiter entry. Missing entries are fine (the acquire that succeeded already removed it).
func (*Store) Enqueue ¶
Enqueue registers w as a waiter for key and returns the entry path, which the caller uses for Heartbeat and Dequeue. Creating a uniquely-named file is atomic, so no guard is needed.
func (*Store) Heartbeat ¶
Heartbeat proves the waiter behind entryPath is still alive by touching its mtime. If the entry vanished (e.g. an aggressive prune), it is re-created so the waiter keeps its queue position.
func (*Store) Prune ¶
func (s *Store) Prune() (*PruneResult, error)
Prune removes expired leases and stale waiter entries. It never removes lock directories: an open guard file descriptor in another process must stay valid, and empty directories are free.
func (*Store) Release ¶
Release drops the lease for key. Without force, token must match the current lease. Returns the released holder (nil if force-releasing corrupt state).
func (*Store) Renew ¶
Renew extends the lease for key by ttl from now. Token must match. A lease that expired but has not been displaced yet is revived — better to keep a live agent's lease than to invite a collision.
func (*Store) Status ¶
Status returns a read-only snapshot of key. It takes no guard: holder.json is replaced atomically, so the read is consistent.
func (*Store) TryAcquire ¶
func (s *Store) TryAcquire(key, token string, o AcquireOpts) (*AcquireResult, error)
TryAcquire makes one attempt to take the lease. It never blocks (beyond the millisecond-scale guard). Pessimistic waiting is the caller's poll loop around this.
type Waiter ¶
type Waiter struct {
Token string `json:"token,omitempty"`
Agent string `json:"agent"`
PID int `json:"pid"`
Host string `json:"host"`
Reason string `json:"reason,omitempty"`
EnqueuedAt time.Time `json:"enqueued_at"`
// Computed on read, not stored.
Fresh bool `json:"fresh"`
LastSeen time.Time `json:"last_seen,omitzero"`
}
Waiter is a queued acquire attempt.
Source Files