Help us shape the future of Go! Take the 2026 Go Developer Survey

leases

package
v0.40.22 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 2, 2026 License: MIT Imports: 11 Imported by: 0

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

View Source 
const DefaultTTL = 2 * time.Minute

DefaultTTL is the lease lifetime stamped on each Acquire/Renew. Pods renew every TTL/4 — see Coordinator.reconcileInterval.

View Source 
const (
	GroupTTL = 30 * time.Second
)

Lead mode timing (design 4.1): a 30 s lease renewed every 5 s (+-1 s of jitter), a 3 s statement timeout on the renew, and a local deadline of 20 s after the renew was sent.

Variables

View Source 
var ErrLeadModeInactive = errors.New("leases: lead mode was requested (RIVOLT_GROUP_LEAD=1) but the lease store is not leading; refusing to run the old per-vehicle mode in its place")

ErrLeadModeInactive: lead mode was requested but this coordinator would not lead.

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.)

func QueryStringColumn 

func QueryStringColumn(ctx context.Context, db *sql.DB, query string) ([]string, error)

QueryStringColumn runs a one-column SELECT and returns every non-empty value. Helper for building DB-backed VehicleSourceFunc closures in main.go without duplicating the boilerplate.

Types

type Activity added in v0.40.13 

type Activity int

Activity is what a vehicle is doing, for the lead-mode shed choice (design 4.7). 

const (
	// ActivityIdle is a vehicle asleep (or with no state yet): moving its group costs nothing.
	ActivityIdle Activity = iota
	// ActivityAwake is awake with no open session.
	ActivityAwake
	// ActivityBusy has an open drive or charge: its group is never shed (prod gate d), and a group
	// change under it is followed, not handed over.
	ActivityBusy
)

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) CheckLeadMode added in v0.40.15 

func (c *Coordinator) CheckLeadMode(requested bool) error

CheckLeadMode records whether the operator asked for lead mode and returns ErrLeadModeInactive if so but the store would not lead: the lead pass skips a store that is not a lead-mode store without a word, and a pod that believes it is in lead mode while it runs the old mode is the worst state to be in. main refuses to start on the error; Run returns it too.

func (*Coordinator) MarkDraining added in v0.40.11 

func (c *Coordinator) MarkDraining(ctx context.Context)

MarkDraining stops this pod taking leases and tells its peers (lease_pods.draining), so they leave it out of the fair share and the newest generation. It keeps renewing what it holds until Shutdown releases it. Safe to call more than once. A failure is logged: the pod still drains locally and the registration is repeated on the next pass.

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) SetActivityCheck added in v0.40.13 

func (c *Coordinator) SetActivityCheck(fn func(vehicleID string) Activity, onShed func(choice string))

SetActivityCheck wires the per-vehicle activity probe the lead-mode shed uses, and an optional observer of each shed decision. Without it the shed takes the store's default (highest id). Safe to call before Run.

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) SetFrameObserver added in v0.40.12 

func (c *Coordinator) SetFrameObserver(frameAt func(vehicleID string) time.Time, observe func(seconds float64, active bool))

SetFrameObserver wires the acquire-to-first-frame measurement: frameAt is when the newest frame for a vehicle arrived on this pod (zero when none), observe is told the seconds from acquire to it.

func (*Coordinator) SetGroupObserver added in v0.40.9 

func (c *Coordinator) SetGroupObserver(held func(n int), dryRun func(source string, 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) SetRenewObserver added in v0.40.15 

func (c *Coordinator) SetRenewObserver(fn func(scope, outcome string))

SetRenewObserver wires the callback told every renew attempt's outcome (scope "vehicle" or "group"; ok, lost, timeout, error). Call before Run.

func (*Coordinator) SetSelfFenceObserver added in v0.40.12 

func (c *Coordinator) SetSelfFenceObserver(fn func())

SetSelfFenceObserver wires the callback fired when the pod stops everything because its group lease renewals failed past the local deadline. 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 GroupLead added in v0.40.11 

type GroupLead struct {
	// Held is the groups this pod holds after the pass, with their epochs.
	Held map[string]int64
	// Lost are groups held last pass that are no longer ours (expired and taken, or no longer a
	// candidate). Their subscriptions must stop; the row is not ours to release.
	Lost []string
	// Shed are groups held beyond the fair share (one per pass). The caller stops them and then
	// calls ReleaseGroup: stop first, release second.
	Shed []string
	// ShedOrder is every held group in the order a shed takes them (highest id first), set when
	// the pod is over its share; Shed is its first. A coordinator that knows which groups are
	// busy (an open drive or charge) picks the first idle one instead (Coordinator.SetActivityCheck).
	ShedOrder []string
	// Retire are held rows that are no longer candidates (the session was removed, or the group was
	// re-keyed). Same rule as Shed: stop first, then ReleaseGroup. They are also in Lost.
	Retire []string
}

GroupLead is what one lead-mode pass decided.

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 

type GroupToken struct {
	Group string
	Epoch int64
}

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 

func NewStore(db *sql.DB, podID string) (*Store, error)

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 

func (s *Store) Acquire(ctx context.Context, vehicleID string) (bool, error)

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) AcquireGroups added in v0.40.13 

func (s *Store) AcquireGroups(ctx context.Context, groups []string) (map[string]int64, error)

AcquireGroups takes the given candidate groups when they are free (no row, expired, or released), outside the fair share and the newest-generation grace, and returns the ones it got with their epochs. The coordinator calls it for a vehicle with an open drive or charge whose group changed under it (re-keyed or split by a shared vehicle), so the session is not handed over mid-way; the next passes shed the group once it is idle. A draining pod acquires nothing. The re-key guard holds as in LeadGroups.

func (*Store) Deadline added in v0.40.12 

func (s *Store) Deadline(vehicleID string) (time.Time, bool)

Deadline is the local lease deadline for the vehicle's group: when the last renew was sent plus the local deadline (monotonic). ok is false with lead mode off; a vehicle whose group this pod does not hold has a deadline in the past.

func (*Store) DropGroups added in v0.40.12 

func (s *Store) DropGroups()

DropGroups forgets every group token (the pod has stopped acting for them). The rows expire on their own.

func (*Store) Fence added in v0.40.12 

func (s *Store) Fence(vehicleID string) (fence.Token, fence.State)

Fence says what this pod may write for a vehicle (fence.Source). Lead mode off: Off. In lead mode a write needs a held group for the vehicle, renewed within the local deadline; a group stolen within the warm-up window waits.

func (*Store) FenceUser added in v0.40.13 

func (s *Store) FenceUser(ctx context.Context, uid string) (fence.Token, fence.State, error)

FenceUser is Fence for a user rather than a vehicle (the receipts sync is per user): the token of the user's group when this pod holds it. One query maps the user to their group (the same userGroups CTE the leases use); the rest is the in-memory state Fence reads. A user with no vehicle has no group, so no owner, and is Lost in lead mode.

func (*Store) GroupLeading added in v0.40.11 

func (s *Store) GroupLeading() bool

GroupLeading reports whether lead mode is on.

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) GroupVehicles added in v0.40.11 

func (s *Store) GroupVehicles(ctx context.Context) (map[string]string, error)

GroupVehicles maps every Rivian vehicle id to the id of its user group.

func (*Store) LeadGroups added in v0.40.11 

func (s *Store) LeadGroups(ctx context.Context) (GroupLead, error)

LeadGroups is one lead-mode pass: register, renew the groups this pod holds, acquire free ones up to the fair share, and report what to shed. Statements per pass: pod register, prune, live pods, candidate count, renew, acquire, release of non-candidates. No new connection.

func (*Store) LocalDeadline added in v0.40.12 

func (s *Store) LocalDeadline() time.Duration

LocalDeadline is how long after a renew was sent this pod may still act: TTL - 10 s (20 s of a 30 s lease). Group and vehicle leases fence on this one rule.

func (*Store) MarkDraining added in v0.40.11 

func (s *Store) MarkDraining(ctx context.Context) error

MarkDraining records that this pod is shutting down: it acquires nothing more and is not counted when the fair share is worked out. A draining pod keeps renewing what it holds until it releases.

func (*Store) PastLocalDeadline added in v0.40.11 

func (s *Store) PastLocalDeadline() bool

PastLocalDeadline reports whether this pod holds groups whose last successful renew was sent more than the local deadline ago: it would be fencing itself (M2) and its writes would be refused. It uses the monotonic clock and no database call. False when no group is held.

func (*Store) PodID 

func (s *Store) PodID() string

PodID returns the identity this Store acquires leases under.

func (*Store) Release 

func (s *Store) Release(ctx context.Context, vehicleID string) error

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 

func (s *Store) ReleaseAll(ctx context.Context) (int, error)

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) ReleaseGroup added in v0.40.11 

func (s *Store) ReleaseGroup(ctx context.Context, group string) error

ReleaseGroup gives one group row back (pod_id NULL, released_at set) after its subscriptions have stopped, and forgets its token.

func (*Store) Renew 

func (s *Store) Renew(ctx context.Context) ([]string, error)

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.

  1. Check the tokens held last pass (dry run: log and count, nothing acts).
  2. 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.
  3. 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) SetGroupLead added in v0.40.11 

func (s *Store) SetGroupLead(on bool)

SetGroupLead turns lead mode on or off. On, the lease TTL is 30 s and the coordinator ticks every 5 s; call it before NewCoordinator.

func (*Store) SetVehicleGroups added in v0.40.12 

func (s *Store) SetVehicleGroups(vg map[string]string)

SetVehicleGroups records which group each vehicle belongs to, as of the last lead pass (the coordinator calls it on every pass that read the map, before it acquires or subscribes anything). Fence and Deadline look vehicles up here: a vehicle with no entry reads Lost, so without this call a pod that holds every group still refuses all of its own writes. The map is copied.

func (*Store) SplitVehicles added in v0.37.6 

func (s *Store) SplitVehicles(ctx context.Context) ([]string, error)

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 

type VehicleSourceFunc func(ctx context.Context) ([]string, error)

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

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.