Documentation
¶
Overview ¶
Package leases coordinates per-vehicle ownership across multiple rivolt pods so each vehicle is subscribed by exactly one process.
Mechanism (see migration 0011):
- The `subscription_leases` table pins a vehicle to a pod_id.
- Acquire() inserts a row, or "steals" one whose expires_at is in the past via INSERT ... ON CONFLICT DO UPDATE WHERE.
- Renew() bumps renewed_at + expires_at on rows we own.
- ReleaseAll() deletes our rows on SIGTERM so a planned restart hands ownership over without waiting for the TTL.
Coordinator is the run loop that ties Store to a callback surface: "tell me when I gain a vehicle, tell me when I lose one." The state-monitor goroutine wires those callbacks to EnsureSubscribed / Unsubscribe so the WS subscription set tracks lease ownership.
Trade-offs:
- Acquisition is opportunistic — every pod tries every vehicle each cycle. At 1000 vehicles / 3-8 pods that's a few hundred INSERTs every 30s, which Postgres eats trivially. Consistent- hashing the vehicle set across pods is a future optimization when the storm becomes visible in pg_stat_statements.
- The TTL has to be long enough that a pod under load doesn't lose its leases mid-renew, and short enough that a crashed pod's vehicles aren't stale for too long. 2 minutes is the architecture-doc default; renew every 30s gives 4x headroom.
Index ¶
- Constants
- func NewVehicleSource(monitor func() []string, logger *slog.Logger, dbSources ...VehicleSourceFunc) func(ctx context.Context) ([]string, error)
- func QueryStringColumn(ctx context.Context, db *sql.DB, query string) ([]string, error)
- type Coordinator
- func (c *Coordinator) Run(ctx context.Context) error
- func (c *Coordinator) SetAuthoritative(fn func(ctx context.Context) ([]string, error))
- func (c *Coordinator) SetCountObserver(fn func(count int))
- func (c *Coordinator) SetGroupObserver(held func(n int), dryRun func(n int))
- func (c *Coordinator) SetStreamingCheck(fn func(vehicleID string) bool)
- func (c *Coordinator) SetUnstreamedObserver(fn func(vehicleID string))
- func (c *Coordinator) Shutdown(ctx context.Context)
- func (c *Coordinator) TriggerReconcile()
- type GroupResult
- type GroupToken
- type Store
- func (s *Store) Acquire(ctx context.Context, vehicleID string) (bool, error)
- func (s *Store) GroupTokens() []GroupToken
- func (s *Store) PodID() string
- func (s *Store) Release(ctx context.Context, vehicleID string) error
- func (s *Store) ReleaseAll(ctx context.Context) (int, error)
- func (s *Store) Renew(ctx context.Context) ([]string, error)
- func (s *Store) RenewGroups(ctx context.Context) (GroupResult, error)
- func (s *Store) SplitVehicles(ctx context.Context) ([]string, error)
- type VehicleSourceFunc
Constants ¶
const DefaultTTL = 2 * time.Minute
DefaultTTL is the lease lifetime stamped on each Acquire/Renew. Pods renew every TTL/4 — see Coordinator.reconcileInterval.
Variables ¶
This section is empty.
Functions ¶
func NewVehicleSource ¶
func NewVehicleSource( monitor func() []string, logger *slog.Logger, dbSources ...VehicleSourceFunc, ) func(ctx context.Context) ([]string, error)
NewVehicleSource composes one in-memory monitor source with any number of DB-backed sources into a single coordinator vehicle-set closure. The returned function dedupes results across sources, drops empty strings, and tolerates errors from individual sources (logged at debug, skipped — the next tick will retry).
`monitor` is the in-process StateMonitor's known set, called fresh on every reconcile so newly-logged-in users on this pod show up without a restart. Pass nil to skip.
`dbSources` are typically built with QueryStringColumn against the vehicles and subscription_leases tables; that combination gives true N>1 awareness while still booting a single pod with an empty database.
Including monitor here is non-negotiable — `vehicles` only gets rows once the recorder writes a sample, and the recorder only runs once a subscription is up. Without monitor, you have a chicken-and-egg that silently records nothing. (See the v0.16.1 regression.)
Types ¶
type Coordinator ¶
type Coordinator struct {
// contains filtered or unexported fields
}
Coordinator runs the lease reconciliation loop and bridges Acquire/lose events to subscription start/stop callbacks.
Lifecycle:
c := NewCoordinator(store, vehicles, onAcquire, onRelease, logger) go c.Run(ctx) // ticks every reconcileInterval defer c.Shutdown(ctx) // calls ReleaseAll
func NewCoordinator ¶
func NewCoordinator( store *Store, vehicles func(ctx context.Context) ([]string, error), onAcquire, onRelease func(vehicleID string), logger *slog.Logger, ) *Coordinator
NewCoordinator builds a coordinator. onAcquire/onRelease must be non-nil; the rest are optional (logger defaults to slog.Default, onCountChange and reconcile-tuning to sane defaults).
func (*Coordinator) Run ¶
func (c *Coordinator) Run(ctx context.Context) error
Run blocks until ctx is cancelled, reconciling leases on every tick. Returns ctx.Err() so callers can distinguish a clean shutdown from an unexpected exit.
Reconciles once immediately so a freshly-started pod doesn't wait reconcileInterval before claiming any vehicles.
func (*Coordinator) SetAuthoritative ¶ added in v0.25.0
func (c *Coordinator) SetAuthoritative(fn func(ctx context.Context) ([]string, error))
SetAuthoritative wires the existence source used to reap orphan leases — vehicles whose backing row was deleted while a lease lingered (subscription_leases has no FK to users, so a deleted tenant's leases survive the cascade). Without it those leases are renewed forever and the owning pod's recorder keeps failing FK inserts. Safe to call before Run.
func (*Coordinator) SetCountObserver ¶
func (c *Coordinator) SetCountObserver(fn func(count int))
SetCountObserver wires a callback that fires after every reconcile with the current number of leases owned. Safe to call before Run.
func (*Coordinator) SetGroupObserver ¶ added in v0.40.9
func (c *Coordinator) SetGroupObserver(held func(n int), dryRun func(n int))
SetGroupObserver wires the metrics for the group leases: held is told the number of groups this pod holds after every pass, dryRun the number of tokens that would have been refused. Optional.
func (*Coordinator) SetStreamingCheck ¶ added in v0.33.0
func (c *Coordinator) SetStreamingCheck(fn func(vehicleID string) bool)
SetStreamingCheck wires the "is this pod subscribed to the vehicle" probe that lets reconcile heal a lease held without a subscription. Safe to call before Run.
func (*Coordinator) SetUnstreamedObserver ¶ added in v0.33.0
func (c *Coordinator) SetUnstreamedObserver(fn func(vehicleID string))
SetUnstreamedObserver wires a callback fired once per episode of a lease held without a subscription (the metric). Safe to call before Run.
func (*Coordinator) Shutdown ¶
func (c *Coordinator) Shutdown(ctx context.Context)
Shutdown drops every lease this pod owns and fires onRelease for each, stopping the subscriptions first. Used from the SIGTERM hook so a planned pod restart hands vehicles to surviving pods immediately. Best-effort with a caller-supplied (timeout-bound) context — if Postgres is unreachable the leases will still age out via the TTL.
func (*Coordinator) TriggerReconcile ¶
func (c *Coordinator) TriggerReconcile()
TriggerReconcile asks the Run loop to reconcile on its next scheduling pass instead of waiting for the next tick. Safe to call concurrently and from any goroutine; multiple calls coalesce into a single reconcile (the channel is cap-1, send is non-blocking).
Used at startup: RefreshVehicleInfo's success goroutine calls this so the first set of vehicles claims its leases within milliseconds of the REST reply, instead of waiting up to a full reconcileInterval.
type GroupResult ¶ added in v0.40.9
type GroupResult struct {
// Held is the number of groups this pod now holds.
Held int
// DryRunRejected lists groups whose token from the previous pass would have
// been refused by rivolt_lease_ok now (the lease expired or moved under us).
DryRunRejected []string
}
GroupResult is what one RenewGroups pass saw.
type GroupToken ¶ added in v0.40.9
GroupToken is the proof a pod holds a group: the lease's epoch changes on every acquire or steal and never on a renew.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the data-access layer for the subscription_leases table. All methods are safe to call concurrently.
func NewStore ¶
NewStore wraps a *sql.DB and pins the calling pod's identity. The pod_id should be stable for the process lifetime — typically the k8s pod name via the downward API. An empty pod_id is a configuration bug; NewStore returns an error rather than letting every pod share an identity (which would silently disable the whole coordination layer).
func (*Store) Acquire ¶
Acquire attempts to claim ownership of vehicleID. Returns true if this pod now owns the lease (newly inserted OR successfully stolen from an expired holder), false if a healthy lease held by another pod blocked the claim.
The single SQL statement does both insert-and-steal in one atomic step so two pods racing on the same expired lease both can't win: Postgres' row-level lock during ON CONFLICT serializes them.
func (*Store) GroupTokens ¶ added in v0.40.9
func (s *Store) GroupTokens() []GroupToken
GroupTokens returns the fence tokens this pod holds, as of the last RenewGroups.
func (*Store) Release ¶
Release drops a single lease. Used when a vehicle disappears from /api/vehicles (e.g. the user removed it) so we don't keep pretending to own a no-longer-existing subscription.
func (*Store) ReleaseAll ¶
ReleaseAll drops every lease owned by this pod. Called from the SIGTERM hook so a graceful shutdown hands ownership to surviving pods on their NEXT reconcile cycle (no waiting for the TTL).
func (*Store) Renew ¶
Renew extends every lease this pod currently owns. Returns the vehicleIDs that are still owned after the renew (i.e. weren't stolen out from under us between the previous reconcile and now).
A vehicle missing from the result set is one we lost ownership of — the caller (Coordinator) translates that into an Unsubscribe callback so the WS subscription is torn down promptly.
func (*Store) RenewGroups ¶ added in v0.40.9
func (s *Store) RenewGroups(ctx context.Context) (GroupResult, error)
RenewGroups writes this pod's group lease rows and registers the pod. It is called once per reconcile, after the vehicle leases are renewed and acquired.
- Check the tokens held last pass (dry run: log and count, nothing acts).
- Upsert a row for every group whose vehicle leases are all mine and unexpired. A row this pod already holds and that has not expired keeps its epoch; any other acquire or steal takes a new one. A row held by another pod is taken only once it has expired.
- Release this pod's rows for groups it no longer fully holds (pod_id NULL, released_at set; the row stays so M2 can tell how long a group has been free).
A group whose users have a stored session but no vehicle gets no row in M1; M2 widens the candidates. Per tick that is 3 statements (pod registration, one query for the tokens held last pass, the upsert) plus one release UPDATE, all on the system pool's existing connections.
func (*Store) SplitVehicles ¶ added in v0.37.6
SplitVehicles lists the vehicles this pod holds whose user also has an unexpired lease on another vehicle held by a pod with a LOWER pod id. Such a user is split across pods; the greater pod id gives its share up, so both pods never release at once and the split converges. The Coordinator releases them (unsubscribe first, then Release).
type VehicleSourceFunc ¶
VehicleSourceFunc yields a snapshot of vehicle IDs for one underlying source. Errors are surfaced to the composer, which logs and falls through to the remaining sources rather than failing the whole reconcile.
Source Files