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) 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 Store
- func (s *Store) Acquire(ctx context.Context, vehicleID string) (bool, error)
- 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) 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) 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 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) 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) 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