forward

package
v0.41.5-rc.3 Latest Latest
Warning

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

Go to latest
Published: Nov 9, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package forward is the single-owner forwarder (design §2/§3, M3): a request that needs Rivian from a pod that does not own the user's group is pushed to the owner through Redis lists and answered there, so only the owner calls Rivian.

This is the M3-0 skeleton: the message shapes, the signing, the per-kind allowlist with strict validators, the Redis key names, the id generator, the expiry arithmetic, the metrics and the injected interfaces with their fakes. Nothing here is wired into the server and nothing talks to Redis or Postgres yet; the owner loop (M3-1), the requester and the kinds follow in later PRs, one each.

Rules that hold for every later step:

  • Tokens, passwords and cookies never ride a message. A message names a user and a group, never a credential. The `adopt` kind carries only user and group.
  • Every request and reply is signed (HMAC-SHA256, C1) with a purpose-labelled subkey of one install secret; the receiver verifies the signature before anything else (before the schema, the lease or the Redis clock). The two directions use different subkeys so a reply can never be replayed as a request.
  • Time is Redis's: `expires_ms` is computed from Redis TIME and compared with Redis TIME; pods never compare wall clocks.
  • Keys live under the app's ACL prefix `rivolt:` (the Valkey user `rivolt` has `~rivolt:*`).
  • Metric labels carry no user, vehicle or group (/metrics is public).

Index

Constants

View Source
const (
	RequestListTTL = 45 // the longest kind timeout plus slack; refreshed on every push
	ReplyTTL       = 5  // the requester is already blocked reading when the owner answers
	UserSetTTL     = 60 // the per-user set of reply ids, for purge on account deletion
	ReadyTTL       = 15 // the owner's readiness marker; refreshed every ReadyEvery
	ReadyEvery     = 5
)

Reply and request lifetimes (seconds).

View Source
const (
	MaxRequestBytes = 64 << 10
	MaxReplyBytes   = 256 << 10
)

Size caps (bytes). A request is one Redis list element; a reply is one element of a short-lived list.

View Source
const (
	StatusMoved      = 421 // this pod no longer holds the group at that epoch: the requester re-reads the lease and retries once
	StatusBusy       = 503 // the user already has the maximum in flight
	StatusInternal   = 500
	StatusTooBig     = 502
	StatusBadRequest = 400 // a message our own key signed that fails the kind's schema: a bug on the sending side, answered at once
	ClassBadRequest  = "bad_request"
	ClassMoved       = "moved"
	ClassBusy        = "busy"
	ClassInternal    = "internal"
	ClassReplyTooBig = "reply_too_big"
)

Statuses an Owner replies with when it does not run the kind. They are HTTP-ish numbers the requester maps with the handlers' own error mapping (the Class travels with them).

View Source
const (
	StatusUnavailable = 503
	ClassUnavailable  = "unavailable"
	StatusBadReply    = 502
	ClassBadReply     = "bad_reply"
	RetryAfterSeconds = 5 // the Retry-After a handler sends with a 503 from here
)

Statuses and classes the requester itself produces (an owner's replies carry their own).

View Source
const DefaultPrefix = "rivolt"

DefaultPrefix is the key prefix. The Valkey user `rivolt` may touch `~rivolt:*` only, so every forwarder key starts with it.

View Source
const EnvKey = "RIVOLT_FORWARD_HMAC_KEY"

EnvKey is the environment variable that carries the key, base64 (32 random bytes: 44 characters). It is optional until the forwarder flag is on. Vault: kv/rivolt/forward and kv/rivolt-preview/forward, property hmac_key.

View Source
const KeyLen = 32

KeyLen is the minimum length of the HMAC key (the install secret), in bytes: a key is valid when it decodes to at least this many.

View Source
const KindUnknown = "unknown"

KindUnknown is the metric label for a message whose kind is not in the allowlist (or that could not be decoded). It is created at boot so a flood of garbage shows as rejected{kind="unknown"} without an unbounded label.

View Source
const MaxQueue = 100

MaxQueue is the owner list length above which a requester refuses to push (503, Retry-After: 5).

View Source
const Version = 1

Version is the message format version. A message with another version is dropped and counted rejected.

Variables

View Source
var (
	// ErrBadKey: the key is missing, not base64, or shorter than KeyLen bytes.
	ErrBadKey = errors.New("forward: the HMAC key must be at least 32 random bytes, base64-encoded")
	// ErrBadSig: the signature is missing, malformed or wrong.
	ErrBadSig = errors.New("forward: bad signature")
	// ErrTooBig: a message over its size cap.
	ErrTooBig = errors.New("forward: message too big")
	// ErrBadMessage: not a message of this version, or a malformed one.
	ErrBadMessage = errors.New("forward: bad message")
)
View Source
var ErrBadArgs = errors.New("forward: bad args")

ErrBadArgs: args that fail the kind's schema.

View Source
var Kinds = map[Kind]Spec{
	KindPlanLegs:   {Timeout: 28 * time.Second, Retry: true, Validate: func(a json.RawMessage) (any, error) { return decodeArgs[PlanLegsArgs](a) }},
	KindFreshState: {Timeout: 10 * time.Second, Retry: true, Validate: func(a json.RawMessage) (any, error) { return decodeArgs[FreshStateArgs](a) }},
	KindAdopt:      {Timeout: 5 * time.Second, Retry: false, Validate: func(a json.RawMessage) (any, error) { return decodeArgs[AdoptArgs](a) }},
	KindOTANotes:   {Timeout: 15 * time.Second, Retry: true, Validate: func(a json.RawMessage) (any, error) { return decodeArgs[OTANotesArgs](a) }},
}

Kinds is the allowlist.

View Source
var Outcomes = []string{"ok", "timeout", "moved", "local", "unavailable", "rejected", "error", "local_busy", "held", "no_group", "too_big", "acquired"}

Outcomes of a forwarded call, all created at 0 at boot (a counter born at its first event is invisible to increase()).

  • ok: the owner answered;
  • timeout: no answer in the kind's time, or the request expired before it ran;
  • moved: the owner no longer holds the group (stale epoch or past its local deadline);
  • local: the call was served on this pod (no owner, owner not ready, flag off, Redis down): the M3 fail-soft path;
  • unavailable: refused (queue full, no ready owner on a pod that holds no session): 503 Retry-After: 5;
  • rejected: a message the owner dropped (bad signature, unknown kind, bad args, wrong size);
  • held: this pod holds the group, so the call ran here with no forwarding (the owner's own calls). Counted so total volume and the held:forwarded split are visible; it is not a fail-soft path and is left out of the local-share ratio;
  • local_busy: the owner answered `busy` twice (the user's concurrency cap) and the call was served on this pod (a pod that cannot serve locally answers unavailable instead);
  • no_group: this pod does not know the vehicle's group (not in the lease store's vehicle->group map), so the call ran here with no forwarding. A fail-soft path like local: a non-owner Rivian call the local-share ratio must see;
  • too_big: the owner ran the kind but its answer does not fit the reply cap (MaxReplyBytes less headroom); the requester gets a reply_too_big error. Counted apart from ok so a failure the user sees is not counted as success;
  • acquired: acquire-on-request (M4-2): the lease was absent or expired, this pod took the group and ran the call as its owner;
  • error: the owner ran the kind and the runner failed with an error that is not a timeout (a 500 internal reply). "ok" means served, so the local-share and health arithmetic do not count a failure as success.

Functions

func Check

func Check(s *Signer, r *Request) (any, error)

Check is the owner's whole admission test for a decoded request, in order: the signature, then the kind, then the args. It returns the decoded args on success. Nothing before the signature touches the message's content.

func Expired

func Expired(redisNowMS, expiresMS int64) bool

Expired reports whether a request's expiry has passed on the Redis clock. Pods never compare wall clocks.

func ExpiresMS

func ExpiresMS(redisNowMS int64, k Kind) int64

ExpiresMS is the Redis-clock expiry of a request of kind: Redis TIME in ms plus the kind's timeout.

func MetricKinds

func MetricKinds() []string

MetricKinds are the kind label values.

func NewID

func NewID() (string, error)

NewID returns a fresh random request id: 128 bits from crypto/rand, never derived from a user, vehicle or time.

func ParseKey

func ParseKey(b64 string) ([]byte, error)

ParseKey decodes the base64 key and checks its length (at least KeyLen bytes after decoding). The value is never logged or returned in an error.

func ValidDriveMode added in v0.41.5

func ValidDriveMode(mode string) bool

ValidDriveMode reports whether a forwarded plan may carry mode.

func ValidID

func ValidID(s string) bool

ValidID reports whether s is a request id: 32 lower-case hex characters (128 bits). The owner builds the reply key from an id it has validated, so a request cannot name an arbitrary key.

func ValidPod

func ValidPod(s string) bool

ValidPod reports whether s can be a pod id inside a key.

Types

type AdoptArgs

type AdoptArgs struct{}

AdoptArgs: nothing. The message's user and group are the whole content, and no credential, token or cookie ever rides it.

type Call

type Call struct {
	Kind  Kind
	User  string
	Group string
	Args  any
}

Call is one operation that needs Rivian: the kind, the user and the group it is for, and the kind's args (a value that marshals to the kind's schema).

type Clock

type Clock interface {
	RedisTimeMS(ctx context.Context) (int64, error)
}

Clock is Redis's clock. Pods never compare wall clocks (design §4.1): the requester stamps `expires_ms` from it and the owner compares against it at dequeue.

type Forwarder

type Forwarder struct {
	// contains filtered or unexported fields
}

Forwarder is the requester side: it decides whether a call that needs Rivian is served here or by the group's owner, and if by the owner pushes it, waits for the signed reply and verifies it. Nothing is acquired (C2): no owner means the call is served here.

func NewForwarder

func NewForwarder(cfg RequesterConfig) (*Forwarder, error)

NewForwarder validates cfg.

func (*Forwarder) Do

func (f *Forwarder) Do(ctx context.Context, c Call, local func(context.Context) (Result, error)) (Result, string, error)

Do runs the call and reports who served it. local is the code the handler runs today on a pod that holds the group; Do calls it when this pod holds the group, and as the fail-soft path when the call cannot be forwarded (outcome `local`), or the owner stays busy (`local_busy`).

Order: held here (no Redis) -> lease row (system pool, cached) -> the owner's readiness marker -> the owner's queue length -> push, wait for the signed reply -> on `moved` or a timeout re-read the lease once and retry if the owner or epoch changed. With nobody new to ask, `moved` is served here (`local`: the owner has just said it is not the owner, which is the no-owner case); a timeout is a 503.

func (*Forwarder) NoGroup

func (f *Forwarder) NoGroup(kind Kind)

NoGroup counts a call the handler served on this pod because it could not tell which group the vehicle belongs to (it never reaches Do).

func (*Forwarder) Nudge

func (f *Forwarder) Nudge(ctx context.Context, user, group string) bool

Nudge is the best-effort `adopt` push: user and group only, no reply is awaited, nothing is retried and nothing is served locally. It reports whether a message was pushed to the group's ready owner.

type FreshStateArgs

type FreshStateArgs struct {
	VehicleID string `json:"vehicle_id"`
	MaxAgeMS  int64  `json:"max_age_ms"`
}

FreshStateArgs: one vehicle-state fetch. MaxAgeMS lets the owner serve its cache when it is fresh enough.

type Keys

type Keys struct{ Prefix string }

Keys builds the Redis key names. The zero value uses DefaultPrefix.

func (Keys) Queue

func (k Keys) Queue(pod string) string

Queue is the owner's request list.

func (Keys) Ready

func (k Keys) Ready(pod string) string

Ready is the owner's readiness marker.

func (Keys) Reply

func (k Keys) Reply(id string) string

Reply is the one-shot reply list of request id. The id must have passed ValidID.

func (Keys) UserSet

func (k Keys) UserSet(user string) string

UserSet is the set of in-flight reply ids of a user (purged with the account).

type Kind

type Kind string

Kind is a forwarded operation. The allowlist is closed: an unknown kind is dropped and counted rejected.

const (
	KindPlanLegs   Kind = "plan_legs"   // one planTrip call (the requester keeps the cache and the quota)
	KindFreshState Kind = "fresh_state" // one vehicle-state fetch
	KindAdopt      Kind = "adopt"       // best-effort nudge after a login: user and group only, never credentials
	KindOTANotes   Kind = "ota_notes"   // one OTA release-notes URL lookup (not stored, not cached by the requester)
)

type LeaseForgetter

type LeaseForgetter interface{ Forget(group string) }

LeaseForgetter is implemented by a LeaseView that caches: Forget drops a group's cached row so the next Owner call reads it fresh.

type LeaseOwner

type LeaseOwner struct {
	Found     bool
	Pod       string
	Epoch     int64
	ExpiresAt time.Time
}

LeaseOwner is what a requester learns about a group's owner from the lease row.

type LeaseView

type LeaseView interface {
	// Owner is the current owner of the group, from the lease row.
	Owner(ctx context.Context, group string) (LeaseOwner, error)
	// Held reports whether this pod holds the group, and at which epoch.
	Held(group string) (epoch int64, ok bool)
	// PastDeadline reports whether this pod's local lease deadline has passed (the pod must stop acting).
	PastDeadline() bool
}

LeaseView is how the forwarder reads the lease. It only reads: no acquire in M3 (C2). The implementation reads the lease row on the SYSTEM pool (C3: row-level security on the user pool could hide the row and read as "no owner"), cached for 2 s, and asks the local lease store about this pod.

type LocalLeases

type LocalLeases interface {
	GroupTokens() []leases.GroupToken
	PastLocalDeadline() bool
}

LocalLeases is what the forwarder asks of this pod's lease store (*leases.Store satisfies it). The forwarder only reads.

type Metrics

type Metrics struct {
	Total      *prometheus.CounterVec   // rivolt_rivian_forward_total{kind,outcome}
	Seconds    *prometheus.HistogramVec // rivolt_rivian_forward_seconds{kind}
	QueueDepth prometheus.Gauge         // rivolt_forward_queue_depth (this pod's owner list, owner loop)
	OwnerReady prometheus.Gauge         // rivolt_forward_owner_ready (1 while this pod's owner loop is running and marked ready)
}

Metrics are the forwarder's instruments. No label carries a user, vehicle or group (/metrics is public).

func NewMetrics

func NewMetrics(reg prometheus.Registerer) *Metrics

NewMetrics creates the instruments, registers them on reg and creates every (kind, outcome) series at 0.

type NetworkPref

type NetworkPref struct {
	NetworkID  string `json:"network_id"`
	Preference int    `json:"preference"`
}

NetworkPref is a charging-network preference.

type OTANotesArgs

type OTANotesArgs struct {
	VehicleID string `json:"vehicle_id"`
	ReleaseID string `json:"release_id,omitempty"`
}

OTANotesArgs: one release-notes URL lookup.

type Owner

type Owner struct {
	// contains filtered or unexported fields
}

Owner is the owner side: one dequeue loop on this pod's list, a bounded worker pool, and the readiness marker. It runs only the kinds in the allowlist, only for groups this pod holds at the request's epoch, and only before the request's Redis-clock expiry.

func NewOwner

func NewOwner(cfg OwnerConfig) (*Owner, error)

NewOwner validates cfg and fills the defaults.

func (*Owner) Run

func (o *Owner) Run(ctx context.Context)

Run blocks until ctx is cancelled. It keeps the readiness marker fresh while it dequeues, and removes the marker on the way out so requesters stop forwarding here at once.

func (*Owner) SetDraining

func (o *Owner) SetDraining()

SetDraining stops advertising this pod as ready (the marker is deleted now and not refreshed). In-flight and queued work finishes.

type OwnerConfig

type OwnerConfig struct {
	Pod     string
	Keys    Keys
	Signer  *Signer
	Redis   redis.Cmdable
	Leases  LeaseView
	Runner  Runner
	Metrics *Metrics // nil: no metrics
	Logger  *slog.Logger

	Workers  int           // concurrent kinds on this pod (default 8)
	PerUser  int           // concurrent kinds for one user (default 2)
	BlockFor time.Duration // BLPOP timeout (default 2 s, so shutdown is prompt)
}

OwnerConfig wires an Owner. Redis is the forwarder's own client (a stuck BLPOP must not starve the rate limiter or the live-state store).

type PGLeaseView

type PGLeaseView struct {
	// contains filtered or unexported fields
}

PGLeaseView is the LeaseView over Postgres and the local lease store. The lease row is read on the SYSTEM pool (C3): row-level security on a user pool could hide the row, and a hidden row reads as "no owner", which is the wrong answer to fail toward.

func NewPGLeaseView

func NewPGLeaseView(db *sql.DB, local LocalLeases) *PGLeaseView

NewPGLeaseView returns a view over db (the system pool) and local (this pod's lease store).

func (*PGLeaseView) Forget

func (v *PGLeaseView) Forget(group string)

Forget drops a group's cached row so the next Owner call reads it fresh (the requester does this before it re-reads after a `moved` reply or a timeout).

func (*PGLeaseView) Held

func (v *PGLeaseView) Held(group string) (int64, bool)

Held reports whether this pod holds the group, and at which epoch, as of its last renew.

func (*PGLeaseView) Owner

func (v *PGLeaseView) Owner(ctx context.Context, group string) (LeaseOwner, error)

Owner reads the group's lease row (cached for leaseCacheTTL). A group with no row, or whose row is released or expired, has no owner (Found false). The pod and epoch of the row are the holder as the database sees it.

func (*PGLeaseView) PastDeadline

func (v *PGLeaseView) PastDeadline() bool

PastDeadline reports whether this pod's local lease deadline has passed.

type PlanLegsArgs

type PlanLegsArgs struct {
	VehicleID               string        `json:"vehicle_id"`
	StartingSoC             float64       `json:"starting_soc"`
	StartingRangeMeters     float64       `json:"starting_range_meters"`
	OriginBearing           float64       `json:"origin_bearing"`
	Waypoints               []Waypoint    `json:"waypoints"`
	TargetArrivalSocPercent *float64      `json:"target_arrival_soc_percent,omitempty"`
	NetworkPreferences      []NetworkPref `json:"network_preferences,omitempty"`
	DriveMode               string        `json:"drive_mode,omitempty"`
	HasAdapter              *bool         `json:"has_adapter,omitempty"`
	RoutePreferences        *RoutePrefs   `json:"route_preferences,omitempty"`
	DepartureTimeMS         int64         `json:"departure_time_ms,omitempty"`
}

PlanLegsArgs mirrors rivian.PlanTripInput (the fields the planner call uses).

type Reply

type Reply struct {
	V      int             `json:"v"`
	ID     string          `json:"id"`
	User   string          `json:"user"`
	OK     bool            `json:"ok"`
	Status int             `json:"status"`
	Class  string          `json:"class,omitempty"`
	Body   json.RawMessage `json:"body,omitempty"`
	Sig    string          `json:"sig"`
}

Reply is what the owner pushes back to the requester's one-shot list. Body is kind-specific JSON; Sig covers every other field.

func DecodeReply

func DecodeReply(b []byte) (*Reply, error)

DecodeReply parses one reply element the same way.

type Request

type Request struct {
	V         int             `json:"v"`
	ID        string          `json:"id"`
	User      string          `json:"user"`
	Group     string          `json:"group"`
	Epoch     int64           `json:"epoch"`
	Kind      Kind            `json:"kind"`
	Args      json.RawMessage `json:"args"`
	ExpiresMS int64           `json:"expires_ms"`
	Sig       string          `json:"sig"`
}

Request is what a requester pushes to the owner's list. Sig covers every other field.

func DecodeRequest

func DecodeRequest(b []byte) (*Request, error)

DecodeRequest parses one list element: size cap, strict JSON (unknown fields rejected), version, id shape. It does not check the signature or the args; the owner verifies the signature first, then calls Spec.Validate.

type RequesterConfig

type RequesterConfig struct {
	Pod     string // this pod (the lease holder name)
	Keys    Keys
	Signer  *Signer
	Redis   redis.Cmdable
	Leases  LeaseView
	Metrics *Metrics
	Logger  *slog.Logger
	// Timeouts overrides a kind's wait (tests). Zero or missing: the kind's own.
	Timeouts map[Kind]time.Duration
	// Jitter is the pause before the one retry after a `busy` reply (default 200-500 ms).
	Jitter func() time.Duration
	// Acquire takes a group whose lease is absent or expired, by the database clock, and reports whether this pod now holds it (single-owner
	// M4-2, the acquire half). Nil, or AcquireOnRequest false: the M3 behaviour, the fail-soft cases run the call here.
	Acquire func(ctx context.Context, group string) (bool, error)
	// AcquireOnRequest (RIVOLT_ACQUIRE_ON_REQUEST) replaces the fail-soft "run it here" with acquire-or-503: the only trigger for an
	// acquire is a lease row absent or expired; a live lease (owner not ready, Redis down, busy, moved to someone) is a 503, never an acquire.
	AcquireOnRequest bool
	// Token returns this pod's fence token for the group, after an acquire (the lease store's view). The call that runs as the new owner carries
	// it in its context, so the owner guard and the permit see the new epoch. Nil: the context is unchanged.
	Token func(group string) (fence.Token, bool)
}

RequesterConfig wires a Forwarder.

type Result

type Result struct {
	Status int
	Class  string
	Body   json.RawMessage
}

Result is what a Runner returns for a kind: the HTTP-ish status, the error class the handlers already map (empty on success) and the kind-specific body.

type RoutePrefs

type RoutePrefs struct {
	AvoidTolls    bool `json:"avoid_tolls"`
	AvoidFerries  bool `json:"avoid_ferries"`
	AvoidHighways bool `json:"avoid_highways"`
}

RoutePrefs are the route-avoidance flags.

type Runner

type Runner interface {
	Run(ctx context.Context, kind Kind, user, group string, args any) (Result, error)
}

Runner executes a kind for a user on this pod: the same code the handler runs today. args is what the kind's Validate returned.

type Signer

type Signer struct {
	// contains filtered or unexported fields
}

Signer signs and verifies messages with two subkeys derived from the install secret, one per direction (domain separation: a reply cannot be replayed as a request, nor the reverse). The install secret itself is not kept.

func NewSigner

func NewSigner(key []byte) (*Signer, error)

NewSigner derives the subkeys from key (at least KeyLen bytes).

func (*Signer) SignReply

func (s *Signer) SignReply(r *Reply)

SignReply sets r.Sig.

func (*Signer) SignRequest

func (s *Signer) SignRequest(r *Request)

SignRequest sets r.Sig.

func (*Signer) VerifyReply

func (s *Signer) VerifyReply(r *Reply) error

VerifyReply checks r.Sig in constant time.

func (*Signer) VerifyRequest

func (s *Signer) VerifyRequest(r *Request) error

VerifyRequest checks r.Sig in constant time. It looks at nothing else about the message.

type Spec

type Spec struct {
	// Timeout bounds the owner's work and the requester's wait. It stays at least 2 s under the handler deadline of every route that
	// forwards the kind (the API's 30 s group deadline; see TestForwardKindTimeoutsFitTheirRoutes), or the requester would be cut
	// off before it could answer or retry.
	Timeout time.Duration
	// Retry says whether a timeout or a `moved` reply may be retried once (reads and version-CAS: yes; adopt: never).
	Retry bool
	// Validate strictly decodes the args (unknown fields rejected) and checks types, lengths and ranges. It returns the decoded
	// value on success.
	Validate func(args json.RawMessage) (any, error)
}

Spec is what the allowlist holds for a kind.

type Validatable

type Validatable interface {
	// contains filtered or unexported methods
}

Validatable is implemented by every args type.

type Waypoint

type Waypoint struct {
	Latitude     float64 `json:"lat"`
	Longitude    float64 `json:"lon"`
	WaypointType string  `json:"type,omitempty"`
	EntityID     string  `json:"entity_id,omitempty"`
}

Waypoint is one stop.

Directories

Path Synopsis
Package forwardtest holds the fakes the forwarder's tests inject: a lease view, a runner that records its calls, and a Redis clock that a test can set apart from the wall clock.
Package forwardtest holds the fakes the forwarder's tests inject: a lease view, a runner that records its calls, and a Redis clock that a test can set apart from the wall clock.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL