Documentation
¶
Overview ¶
Package samples persists raw vehicle_state polling snapshots.
Index ¶
- Constants
- Variables
- func ParseRetentionMonths(v string) (int, error)
- func ParseThinAfterDays(v string) (int, error)
- func RetentionCutoff(now time.Time, months int) time.Time
- func ThinCandidate(months []time.Time, thinned map[time.Time]bool, now time.Time, afterDays int) (time.Time, bool)
- func ThinCandidates(months []time.Time, thinned map[time.Time]bool, now time.Time, afterDays int) []time.Time
- func TimeInWindow(t, now time.Time) bool
- type ArchiveRows
- type DayActivity
- type Downsampled
- type DriveCondition
- type Factory
- type PartitionJanitor
- type Retention
- type Sample
- type Store
- func (s *Store) Close() error
- func (s *Store) Count(ctx context.Context) (int, error)
- func (s *Store) DriveConditions(ctx context.Context, vehicleID string, since, until time.Time) ([]DriveCondition, error)
- func (s *Store) EachAll(ctx context.Context, fn func(Sample) error) error
- func (s *Store) EachAllArchived(ctx context.Context, src ArchiveRows, fn func(Sample) error) (reduced []string, err error)
- func (s *Store) InsertBatch(ctx context.Context, batch []Sample) error
- func (s *Store) InsertBulk(ctx context.Context, batch []Sample) error
- func (s *Store) ListAll(ctx context.Context) ([]Sample, error)
- func (s *Store) ListBetween(ctx context.Context, since, until time.Time, limit int) ([]Sample, error)
- func (s *Store) ListBetweenDownsampled(ctx context.Context, since, until time.Time, limit int) (Downsampled, error)
- func (s *Store) ListBetweenDownsampledVehicle(ctx context.Context, vehicleID string, since, until time.Time, limit int) (Downsampled, error)
- func (s *Store) ListBetweenVehicle(ctx context.Context, vehicleID string, since, until time.Time, limit int) ([]Sample, error)
- func (s *Store) ListSince(ctx context.Context, since time.Time, limit int) ([]Sample, error)
- func (s *Store) Reset(ctx context.Context) (int64, error)
- func (s *Store) SleepActivity(ctx context.Context, vehicleID string, since, until time.Time) ([]DayActivity, error)
- type ThinReport
- type Thinner
Constants ¶
const FullKeep = 7 * 24 * time.Hour
FullKeep is how long a thinned month's full partition stays (detached) so a bad thinning can be undone.
const MaxBatchMonths = 120
MaxBatchMonths is the most distinct months one batch may span (WindowStart to now is about 100).
const MaxDownsampleSpan = 7 * 24 * time.Hour
MaxDownsampleSpan bounds the window that is bucketed. The count and the bucketing scan every row in the window, so a caller-controlled span (anonymous demo callers included) must stay bounded; wider windows, and windows with no upper bound, keep the old cheap behaviour (the oldest `limit` rows, LIMIT stops the scan early).
const MinRetentionMonths = 3
MinRetentionMonths is the shortest retention accepted: anything shorter is almost certainly a typo, and dropping last month's samples by mistake can't be undone.
const MinThinAfterDays = 30
MinThinAfterDays is the youngest data thinning may touch: recent detail is what repairs and plan vs actual look at.
Variables ¶
var ( // WindowStart is the earliest sample time accepted: before Rivian's first // deliveries and before any tool that could export data. WindowStart = time.Date(2018, 1, 1, 0, 0, 0, 0, time.UTC) // WindowAhead is how far past now a sample may be dated (clock skew). WindowAhead = 24 * time.Hour )
Which sample times the store accepts. vehicle_state is partitioned by month and writing a sample creates the month's partition, so a timestamp from the year 1 or 9999 would create a partition for it (an import with scattered dates could create thousands, each taking a lock on the parent). Every writer goes through prepareBatch, which refuses samples outside the window and batches that span too many months; the SQL helper has the same bounds.
var ( // ErrSampleOutOfWindow reports a sample dated before WindowStart or more than // WindowAhead after now. ErrSampleOutOfWindow = errors.New("samples: a sample is dated outside the accepted window") // ErrTooManyMonths reports a batch spanning more than MaxBatchMonths months. ErrTooManyMonths = errors.New("samples: batch spans too many months") )
var ErrNotArchived = errors.New("month not archived: needs a verified archive first")
ErrNotArchived refuses a thinning or a drop that would lose detail no archive holds. Under N19 a month is reduced only after every (user, vehicle) in it has a verified archive object (archive.MonthArchived). Dry runs don't need one.
var ErrThinVerify = errors.New("thinned copy failed verification")
ErrThinVerify reports a thinned copy that failed its checks; nothing was swapped.
Functions ¶
func ParseRetentionMonths ¶ added in v0.37.6
ParseRetentionMonths reads RIVOLT_SAMPLE_RETENTION_MONTHS: "" or "0" keeps samples forever (the default, so a self-hosted install keeps its history unless it opts in); otherwise at least MinRetentionMonths.
func ParseThinAfterDays ¶ added in v0.37.6
ParseThinAfterDays reads RIVOLT_SAMPLE_THIN_AFTER_DAYS: "" or "0" never thins (the default); otherwise at least MinThinAfterDays.
func RetentionCutoff ¶ added in v0.37.6
RetentionCutoff is the first month whose partition is kept: partitions for months before it hold only rows older than months months, and go. Samples therefore live at least months months and less than months+1.
func ThinCandidate ¶ added in v0.37.6
func ThinCandidate(months []time.Time, thinned map[time.Time]bool, now time.Time, afterDays int) (time.Time, bool)
ThinCandidate picks the month to thin: the oldest month whose partition ends at or before now-afterDays and that is not yet thinned. months must be sorted oldest first.
func ThinCandidates ¶ added in v0.38.0
func ThinCandidates(months []time.Time, thinned map[time.Time]bool, now time.Time, afterDays int) []time.Time
ThinCandidates lists every month ThinCandidate would pick in turn, oldest first.
func TimeInWindow ¶ added in v0.37.5
TimeInWindow reports whether t is a sample time the store accepts at now.
Types ¶
type ArchiveRows ¶ added in v0.37.7
type ArchiveRows interface {
Objects(ctx context.Context, uid uuid.UUID) ([]archive.ObjectRef, error)
EachRow(ctx context.Context, uid, vehicleID uuid.UUID, month time.Time, fn func(archive.Row) error) error
}
ArchiveRows is what the backup reads the telemetry archive through (archive.Reader).
type DayActivity ¶ added in v0.30.0
type DayActivity struct {
Day time.Time `json:"day"`
AsleepH float64 `json:"asleep_h"`
IdleAwakeH float64 `json:"idle_awake_h"`
// Sleeps counts the times the car fell asleep that day;
// FailedSleeps the ones that ended within a minute. A healthy car
// sleeps a few times a day for hours; a high failed share means
// something keeps waking it, which is usually what drives
// IdleAwakeH up.
Sleeps int `json:"sleeps"`
FailedSleeps int `json:"failed_sleeps"`
}
DayActivity is one calendar day's (UTC) power-state breakdown, in hours, derived from the persisted power_state column (migration 0038). IdleAwakeH is the interesting one: the car awake while parked and not actively charging — i.e. staying awake for no obvious reason, the vampire-drain signal. Driving and active charging are excluded (both are expected awake time). Only intervals with a recorded power_state contribute, so the series fills forward from the deploy.
type Downsampled ¶ added in v0.33.0
type Downsampled struct {
Samples []Sample
// Reduced is true when the window held more than limit rows and the
// series is a bucketed reduction of it.
Reduced bool
// Total is how many rows the window holds. Only known for a window that was
// counted (bounded, within MaxDownsampleSpan); otherwise it is len(Samples).
Total int
// Truncated is true when the window was NOT bucketed (unbounded, or wider than
// MaxDownsampleSpan) and the result reached the cap: the rows are the OLDEST
// `limit` and the tail may be missing. It can be a false alarm when the window
// holds exactly `limit` rows.
Truncated bool
}
Downsampled describes the result of ListBetweenDownsampled.
type DriveCondition ¶ added in v0.36.10
type DriveCondition struct {
ID string
VehicleID string
StartedAt time.Time
EndedAt time.Time
CreatedAt time.Time
Source string
DistanceMi float64
EnergyKWh float64
AvgSpeedMph float64
OutsideTempC *float64
DriveMode string
}
DriveCondition is a recorded drive with the conditions it was driven in, from the drive's own samples: the median non-zero outside temperature (zero is the live merge's "no reading" sentinel, as in the drive recap) and the drive mode with the most samples.
type Factory ¶
type Factory struct {
// contains filtered or unexported fields
}
Factory hands out per-user *Store instances. See drives.Factory for the rationale.
func NewFactory ¶
func NewFactory(d *sql.DB, resolvers *db.VehicleResolverFactory) *Factory
NewFactory wraps the shared pool + resolver factory.
type PartitionJanitor ¶
type PartitionJanitor struct {
// contains filtered or unexported fields
}
PartitionJanitor keeps `vehicle_state` ahead of the clock by at least `lookaheadMonths` partitions. It does this by calling the `rivolt_ensure_vehicle_state_partition(ts)` SQL helper once per target month.
func NewPartitionJanitor ¶
func NewPartitionJanitor(db *sql.DB) *PartitionJanitor
NewPartitionJanitor wires a janitor with sensible defaults. 3 months of lookahead is enough that a pod that runs continuously for a month still never writes into an unpartitioned range, and a pod that restarts hourly (deploys, node drains) has slack for clock skew. The interval is also 1 hour — cheap, boring, and avoids the "I deployed at 23:59 UTC on the last of the month and the janitor fired at 23:30" edge.
func (*PartitionJanitor) EnsureLookahead ¶
func (j *PartitionJanitor) EnsureLookahead(ctx context.Context) error
EnsureLookahead creates the current partition plus `lookaheadMonths` future partitions if they don't already exist. Idempotent; safe to call repeatedly. Returns the first error encountered but always attempts every month.
func (*PartitionJanitor) Run ¶
func (j *PartitionJanitor) Run(ctx context.Context)
Run blocks until ctx is cancelled, calling EnsureLookahead on an `interval` ticker. Intended to be launched as a dedicated goroutine at startup. Errors are logged, never returned — the next tick will retry.
type Retention ¶ added in v0.37.6
type Retention struct {
// Dropped, when set, is told each partition dropped with its row count
// estimate (pg_class.reltuples): the caller logs and counts it.
Dropped func(name string, estRows int64)
// contains filtered or unexported fields
}
Retention drops vehicle_state partitions past the retention period.
func NewRetention ¶ added in v0.37.6
NewRetention returns the sweep for months of retention, or nil for 0 (keep forever).
type Sample ¶
type Sample struct {
VehicleID string
At time.Time
BatteryLevelPct float64
RangeMi float64
OdometerMi float64
Lat, Lon float64
// LocationFixAt is the timestamp Rivian's GNSS module reported on
// the (Lat, Lon) fix itself, NOT the wall-clock time the recorder
// observed it. Differs from At when the modem loses sky view and
// keeps replaying the last known fix — see migration 0013 for
// the field rationale and the Fort Stockton incident that
// motivated it. Pointer (rather than time.Time + omitempty)
// because Go's encoding/json doesn't honor omitempty on struct
// types: a zero time.Time would serialize as "0001-01-01T00:00:00Z"
// and the frontend would compute a 2000-year fix age. nil
// renders as "fix age unknown" upstream.
LocationFixAt *time.Time `json:",omitempty"`
SpeedMph float64
ShiftState string
DriveMode string `json:"drive_mode,omitempty"` // "everyday", "sport", "all-terrain", etc.; nullable in DB for legacy samples
ChargingState string
// PowerState is the car's power/sleep state ("sleep" | "ready" |
// "standby" | "go"). Empty on legacy rows / imports (NULL in DB).
// Feeds the sleep/activity aggregation.
PowerState string `json:"power_state,omitempty"`
ChargerPowerKW float64
// ChargerStatus is Rivian's chargerStatus, the plug indicator
// (migration 0053); "" when not reported.
ChargerStatus string `json:"charger_status,omitempty"`
ChargeLimitPct float64
InsideTempC float64
OutsideTempC float64
DriveNumber int64
ChargeNumber int64
Source string // "live" | "electrafi_import"
// AltitudeM is the elevation above sea level (meters) at
// (Lat, Lon), looked up by the live recorder against the
// Mapzen Terrarium DEM at sample-write time. Pointer because
// it's nullable: NULL on legacy rows written before migration
// 0019, on samples without a GPS fix, on ElectraFi imports,
// and on cold-cache misses where the tile fetch was kicked
// off async (the next sample on that tile will hit the cache).
AltitudeM *float64 `json:"altitude_m,omitempty"`
// TirePressureMinBar is the minimum of the four corner pressures
// at sample time, in bar. Pointer because nullable: NULL on
// legacy rows pre-migration 0024, ElectraFi imports, and live
// rows where Rivian reported zero on all four corners (sensor
// outage / vehicle parked w/ no recent reading). The efficiency
// analyzer pulls the median across a drive's samples; min-of-4
// keeps the worst tire (the one that actually hurts efficiency)
// dominant without bloating the schema with all four corners.
TirePressureMinBar *float64 `json:"tire_pressure_min_bar,omitempty"`
// PackTempAvg/Max/MinC are high-voltage battery cell temperatures
// (°C), merged in from the Parallax energy.high_voltage.battery_state
// topic. Pointers because nullable: NULL on legacy rows pre-migration
// 0036, imports, and live rows recorded before the vehicle pushed a
// battery_state frame. Avg/max/min kept separately so the spread
// (thermal imbalance) is queryable.
PackTempAvgC *float64 `json:"pack_temp_avg_c,omitempty"`
PackTempMaxC *float64 `json:"pack_temp_max_c,omitempty"`
PackTempMinC *float64 `json:"pack_temp_min_c,omitempty"`
}
Sample is a single polling snapshot. VehicleID is the Rivian gateway string id ("01-242521064"); the store resolves it to the internal vehicles.id UUID on write.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store wraps the vehicle_state table, scoped to one user.
func (*Store) DriveConditions ¶ added in v0.36.10
func (s *Store) DriveConditions(ctx context.Context, vehicleID string, since, until time.Time) ([]DriveCondition, error)
DriveConditions lists the user's drives that started in [since, until) (optionally one vehicle, by Rivian id) with their conditions. Drives under a minute are left out, as in the drive list. Each drive reads only its own window of vehicle_state (primary key vehicle_id, at).
func (*Store) EachAll ¶ added in v0.36.17
ListAll returns every sample for this user, oldest first. Used by the Settings → Backup endpoint. Unbounded — a year of 60 s polls is on the order of 500k rows, ~100 MB JSON; acceptable for a full backup but NOT suitable for UI consumption.
func (*Store) EachAllArchived ¶ added in v0.37.7
func (s *Store) EachAllArchived(ctx context.Context, src ArchiveRows, fn func(Sample) error) (reduced []string, err error)
EachAllArchived is EachAll with full detail for thinned months (N19): a month thinning has reduced is read from the user's archive objects, plus any of the month's rows in Postgres an object lacks (rows that came after it was archived, needs_rearchive or not). A thinned month with a vehicle that has no object, or whose object is missing from the store, falls back to the reduced rows in Postgres and is returned in reduced ("2026-05"), for the backup to say so. Months go oldest first; inside a thinned month, vehicle by vehicle, each by at. With src nil it is EachAll.
An archive object that fails after some of its rows were passed to fn (corrupt, truncated) is an error: the caller must fail the whole backup, not end it as if complete.
func (*Store) InsertBulk ¶ added in v0.32.0
InsertBulk is InsertBatch for large batches: each chunk of up to bulkInsertRows samples is one multi-row INSERT (one round trip) instead of one EXEC per row. Same columns, same ON CONFLICT DO NOTHING, same partition and vehicle handling. Chunks commit independently; a failure returns the error and leaves earlier chunks written, which is safe because inserts are idempotent. Used by replay catch-up; the live recorder keeps InsertBatch.
func (*Store) ListBetween ¶
func (s *Store) ListBetween(ctx context.Context, since, until time.Time, limit int) ([]Sample, error)
ListBetween returns samples in (since, until], oldest first. Zero `until` means "no upper bound" (back-compat with ListSince). The upper bound matters for the drive detail page: without it, opening any drive ages back pulls every sample from the drive's start through now, which can be days of charging+parked rows.
func (*Store) ListBetweenDownsampled ¶ added in v0.33.0
func (s *Store) ListBetweenDownsampled(ctx context.Context, since, until time.Time, limit int) (Downsampled, error)
ListBetweenDownsampled is ListBetween with the cap handled by bucketing instead of truncation. At or under the cap it returns every row exactly as ListBetween does. Without an upper bound, or over MaxDownsampleSpan, it IS ListBetween (never flagged Reduced).
func (*Store) ListBetweenDownsampledVehicle ¶ added in v0.33.2
func (s *Store) ListBetweenDownsampledVehicle(ctx context.Context, vehicleID string, since, until time.Time, limit int) (Downsampled, error)
ListBetweenDownsampledVehicle is ListBetweenDownsampled scoped to one vehicle (Rivian vehicle id); empty means every vehicle of the user. The total and the bucketing count only that vehicle's rows.
func (*Store) ListBetweenVehicle ¶ added in v0.33.2
func (s *Store) ListBetweenVehicle(ctx context.Context, vehicleID string, since, until time.Time, limit int) ([]Sample, error)
ListBetweenVehicle is ListBetween scoped to one vehicle (Rivian vehicle id). An empty vehicleID means every vehicle of the user; an id the user doesn't own matches nothing, so the result is empty, never another user's rows.
func (*Store) ListSince ¶
ListSince returns samples newer than since, up to limit, oldest first.
Limit semantics:
- <= 0 falls back to a reasonable default (1000).
- Anything above the hard cap is clamped DOWN to the cap, not reset to the default. The previous "if limit > 10000 { limit = 1000 }" branch was a foot-gun: the UI asks for 20k samples so a 90-minute drive (~1.7k rows at the WS cadence) returns fully, but the store silently truncated to 1k → the route polyline ended halfway through the trip and the user reported "no GPS coordinates for the end of the route".
func (*Store) Reset ¶
Reset deletes every raw sample (vehicle_state row) for this store's user. See drives.Store.Reset.
func (*Store) SleepActivity ¶ added in v0.30.0
func (s *Store) SleepActivity(ctx context.Context, vehicleID string, since, until time.Time) ([]DayActivity, error)
SleepActivity attributes each inter-sample interval to the vehicle's state at its start and sums per UTC day over [since, until). Intervals are capped at 20 min so a data outage never counts as one giant block — steady-state cadence stays well under that even while the car sleeps (the Rivian cloud replays keep the sample stream dense). vehicleID scopes to one car; empty aggregates all of the user's vehicles.
idle_awake = awake (power_state<>'sleep') AND parked (gear not D/R/N) AND not actively charging (charging_state not charging_active / charging_connecting). An unknown (NULL) gear or charging state counts as parked and not charging, as in internal/drain. That isolates unexplained awake time from normal driving / charging.
Charging is judged by charging_state alone, not charger_power_kw: until the recorder fix alongside this, the stored power kept the last session's value on every parked sample (see dropIdleChargerPower in internal/rivian), which made every awake hour look like charging and flattened idle-awake to zero. charging_state was always right, so keying on it corrects the history as well as new rows.
type ThinReport ¶ added in v0.37.6
type ThinReport struct {
Partition string
RowsBefore, RowsAfter int64
BytesBefore, BytesAfter int64
DryRun bool
}
ThinReport is one month's thinning.
type Thinner ¶ added in v0.37.6
type Thinner struct {
// DryRun computes and reports what a run would keep, and changes nothing.
DryRun bool
// Now is the clock (tests set it).
Now func() time.Time
// Report, when set, is told each month processed: in a dry run what it
// would keep, otherwise what it kept.
Report func(r ThinReport)
// OnRethin, when set, is told each restored (user, vehicle) month thinned
// again once its hold (archive_months.rethin_after) ran out.
OnRethin func(partition string, uid, vid uuid.UUID, deleted int64)
// OnRepair, when set, is told how many objects were flagged for
// re-archiving because their month's rows no longer matched them.
OnRepair func(partition string, flagged int64)
// contains filtered or unexported fields
}
Thinner thins old vehicle_state partitions.
func NewThinner ¶ added in v0.37.6
NewThinner returns the thinning job for afterDays, or nil for 0 (off).
Source Files