internal/server/store/store.go
Ref: Size: 77.1 KiB History
// Package store is the server's durable control-plane state, backed by SQLite:
// the host registry, enrollment tokens, desired VM specs, and freed CIDRs. It
// owns what the fleet should be (desired state); the live actual state reported
// by agents is held in memory by package registry.
package store
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/netip"
"os"
"path/filepath"
"strconv"
"time"
"github.com/a73x/eitri/internal/random"
"github.com/a73x/eitri/internal/transport"
sqlite "modernc.org/sqlite"
)
var ErrNameTaken = errors.New("vm name already in use")
// ErrHostNotFound is returned by CreateVM when the host_id does not exist.
var ErrHostNotFound = errors.New("host not found")
var ErrHostNotEnrolled = errors.New("host not accepting new VMs")
// ErrHostHoldsVolumes is returned by RemoveHost while volume rows remain on the
// host. It is a distinct error because it is a STUCK decommission, not a slow
// one: VMs drain themselves, volumes do not — a volume outlives every guest by
// design, so nothing releases it until its claim is deleted. The graceful sweep
// would otherwise retry forever in silence. Wrapped with the count.
var ErrHostHoldsVolumes = errors.New("host still holds volumes")
// SystemTenant is the audit scope for events with no resolvable tenant — a
// denied enroll attempt, a reap ack racing its host's deletion. It is NOT a
// tenant: no tenants row exists for it, no principal can ever hold it (JIT
// allocation skips the handle), so system rows are durable in the audit_log
// but invisible to every tenant-scoped read. Only AppendAudit callers may
// reference it.
const SystemTenant = "system"
type Store struct {
db *sql.DB
dbDir string
}
type Host struct {
ID, Name, OS, Arch, Provisioner, BridgeCIDR, Status string
// CredGeneration is the host's current credential generation. Credentials
// minted at an older generation are rejected — bumping it revokes that
// one host's outstanding credential without rotating the fleet secret.
CredGeneration int64
EnrolledAt time.Time
// Tenant is the owning tenant (partition key); derived from the enroll token.
Tenant string
// Host OS facts, best-effort, refreshed from the agent's Hello each connect.
OSID, OSPretty, OSVersion, Kernel, CPUModel, Virt string
// UplinkAddr is the address this host presents on the network it reaches
// the control plane over — the address an operator dials to reach a
// published guest port. Empty means the host has not said, never that it
// has none.
UplinkAddr string
}
// HostFacts is the write-side shape for UpdateHostFacts (store stays pb-free;
// syncsvc maps pb.HostFacts → store.HostFacts).
type HostFacts struct {
OSID, OSPretty, OSVersion, Kernel, CPUModel, Virt string
// Provisioner is the backend this agent actually runs guests through. It is
// recorded at enrollment and refreshed here because it can CHANGE under a
// host: a Mac enrolled before its backend existed keeps claiming the one it
// enrolled with, naming a backend that may no longer exist in the tree.
Provisioner string
}
type VM struct {
ID, HostID, Name, ImageURL, ImageSHA256, CloudInit string
VCPUs, MemMB, DiskGB int64
PowerState, Status, LastError, AssignedIP string
// Network is the named host network this VM attaches to, as recorded at
// create — the same name a host advertised in its Hello. '' is the NAT
// underlay: every VM created before this column existed, and any VM
// created without naming one, backfills to it.
Network string
// NetworkIP is the address the site's DHCP server granted this guest on
// its named-network NIC, as its host snooped it. AssignedIP is the other
// address the same guest has, on the host's private fabric — every VM has
// that one, and only a VM with a Network can have this one. '' means not
// discovered (yet, or ever: a guest configured static never asks).
NetworkIP string
SSHAuthorizedKey string
// SSHHostPubKey is the public half of the guest's SSH host key, as its host
// reported it (authorized_keys form). The private half lives on that host
// and the control plane never sees it. SSHHostCert is the certificate the
// control plane signed for it, for the principal it derived from this row.
SSHHostPubKey, SSHHostCert string
// InjectedKey* describe the authorized key eitri installed at create: its
// type, SHA256 fingerprint and comment. They are a RECORD of what eitri
// did, not an input to it — SSHAuthorizedKey is cleared when the key is
// merged into user-supplied cloud-init, and these are not. Keys a user
// hides inside their own cloud_init are their business and are deliberately
// not tracked. Empty on rows created before the columns existed, which is
// honest: the record was never taken.
InjectedKeyType, InjectedKeyFP, InjectedKeyComment string
// TrustedCAs is the tenant user-CA set frozen onto this VM at create — the
// set its guest's sshd will trust, and the answer to "can the certificate I
// am holding open this guest". It is the DESIRED state the agent bakes, not
// a report from the guest: the agent is handed exactly this set and writes
// it into TrustedUserCAKeys, and nothing rewrites it afterwards.
//
// nil means NO record was taken — the row predates the column. An empty
// non-nil set cannot occur: create refuses a tenant that has registered no
// CA, so every VM created since this column existed has at least one. That
// makes nil unambiguously "unrecorded" rather than "trusts nothing", which
// is the distinction every reader of this field depends on.
TrustedCAs []TrustedCA
// VolumeClaimIDs is INPUT TO CreateVM ONLY: the claims this VM asks to
// mount, already resolved from names to ids by the API. No read ever fills
// it — the attachment table is the record, and VolumeIDs is how it reads
// back.
VolumeClaimIDs []string
// VolumeIDs is the read side: the volumes attached to this VM, in the order
// it named their claims, which is the order the guest sees the devices in.
VolumeIDs []string
// The address the gate and every exposure target is AssignedIP.
CreatedAt time.Time
DeletedAt *time.Time
// Tenant is the owning tenant, always derived from the host's — never client-set.
Tenant string
}
// TrustedCA is one user CA as it was frozen onto a VM at create: the label its
// tenant gave it, its SHA256 fingerprint in OpenSSH's spelling, and the
// canonical authorized_keys line itself.
//
// AuthorizedKey is here because the row has to SERVE the trust, not just
// describe it — the snapshot the agent bakes from carries key material, and a
// fingerprint cannot be turned back into a key. Label and Fingerprint are what
// the API shows; AuthorizedKey stays server-side, not because it is secret (a
// CA public key is public by construction) but because it is not the fact a
// reader of the VM object is asking for.
//
// The set is stored as JSON in one column rather than a side table: it is
// immutable once written, only ever read whole, and belongs to the VM's
// lifetime, so a table would add a join and a delete cascade to buy nothing.
type TrustedCA struct {
Label string `json:"label"`
Fingerprint string `json:"fingerprint"`
AuthorizedKey string `json:"authorized_key"`
}
const schema = `
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
-- tenant ids MUST remain DOT-FREE: the jump gate's connect name is
-- <tenant>.<name> and it splits on the FIRST dot (see sshgate/gate.go), while VM
-- names are RFC1123 labels (also dot-free). A dotted tenant id would make the
-- split ambiguous. CreateTenantForIdentity is the only path that mints one and
-- is where this is enforced: it derives the handle from an email with dots
-- flattened.
CREATE TABLE IF NOT EXISTS tenants (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
created_at DATETIME NOT NULL
);
CREATE TABLE IF NOT EXISTS hosts (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
os TEXT NOT NULL,
arch TEXT NOT NULL,
provisioner TEXT NOT NULL,
bridge_cidr TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'enrolled',
enrolled_at DATETIME NOT NULL,
cred_generation INTEGER NOT NULL DEFAULT 1,
tenant TEXT NOT NULL DEFAULT 'default' REFERENCES tenants(id),
os_id TEXT NOT NULL DEFAULT '',
os_pretty TEXT NOT NULL DEFAULT '',
os_version TEXT NOT NULL DEFAULT '',
kernel TEXT NOT NULL DEFAULT '',
cpu_model TEXT NOT NULL DEFAULT '',
virt TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS enrollment_tokens (
token_hash TEXT PRIMARY KEY,
expires_at DATETIME NOT NULL,
used_at DATETIME,
tenant TEXT NOT NULL DEFAULT 'default'
);
CREATE TABLE IF NOT EXISTS vms (
id TEXT PRIMARY KEY,
host_id TEXT NOT NULL REFERENCES hosts(id),
name TEXT NOT NULL,
image_url TEXT NOT NULL,
image_sha256 TEXT NOT NULL,
cloud_init TEXT NOT NULL DEFAULT '',
ssh_authorized_key TEXT NOT NULL DEFAULT '',
ssh_host_cert TEXT NOT NULL DEFAULT '',
vcpus INTEGER NOT NULL,
mem_mb INTEGER NOT NULL,
disk_gb INTEGER NOT NULL,
power_state TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
last_error TEXT NOT NULL DEFAULT '',
assigned_ip TEXT NOT NULL DEFAULT '',
created_at DATETIME NOT NULL,
deleted_at DATETIME,
tenant TEXT NOT NULL DEFAULT 'default' REFERENCES tenants(id)
);
CREATE UNIQUE INDEX IF NOT EXISTS vms_tenant_name ON vms(tenant, name) WHERE deleted_at IS NULL;
-- revoked SSH user certs: a tenant can revoke a specific user cert by its serial
-- (crypto-random uint64, set at mint) so it is rejected at the jump gate before
-- its short TTL expires. serial is stored as the int64 bit-pattern of the uint64
-- (SQLite INTEGER is signed 64-bit) — a bijection, so PRIMARY KEY uniqueness and
-- lookups are preserved. The tenant column (added via ensureColumn in Open — the
-- table shipped without it) scopes the revocation LIST per tenant and records
-- which tenant filed the revocation; old rows backfill to 'default'. Enforcement
-- at the GATE is scoped to the revoking tenant: the gate resolves a presented
-- certificate to the tenant that registered its signing CA, and consults that
-- tenant's rows only. A fleet-wide kill by serial was fail-safe for the guest it
-- protected and an availability hole for everyone else — any tenant could write
-- a row that denied a serial it had no claim to. The key is (tenant, serial) so
-- two tenants can revoke the same serial independently; keyed on serial alone
-- the second revocation would be dropped and its certificate keep working.
-- Guests trust the CA with no guest-side KRL — a multi-user/rotation follow-up.
CREATE TABLE IF NOT EXISTS revoked_ssh_certs (
tenant TEXT NOT NULL DEFAULT 'default',
serial INTEGER NOT NULL,
revoked_at DATETIME NOT NULL,
reason TEXT NOT NULL DEFAULT '',
PRIMARY KEY (tenant, serial)
);
-- tenant_user_cas: uploaded per-tenant USER CA public keys (BYO). eitri holds
-- NO user signing key; it only registers pubkeys. A VM COPIES its tenant's set
-- onto its own row at create (vms.trusted_cas) and the agent bakes that copy
-- into TrustedUserCAKeys, so editing this table never reaches an existing
-- guest; the gate trusts this set (mutable, live)
-- and stamps a connection's tenant from WHICH ca_pubkey verified the cert.
-- ca_pubkey is the canonical authorized_keys line ("type base64", no comment /
-- trailing newline — see sshca.AuthorizedKeyLine). scope is 'tenant' in v1
-- (per-user/fleet later). PRIMARY KEY(tenant, ca_pubkey) makes re-upload a
-- no-op. ca_pubkey is also UNIQUE across tenants so the gate lookup is
-- unambiguous.
CREATE TABLE IF NOT EXISTS tenant_user_cas (
tenant TEXT NOT NULL,
ca_pubkey TEXT NOT NULL,
scope TEXT NOT NULL DEFAULT 'tenant',
label TEXT NOT NULL DEFAULT '',
added_by TEXT NOT NULL DEFAULT '',
created_at DATETIME NOT NULL,
PRIMARY KEY (tenant, ca_pubkey)
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_tenant_user_cas_pubkey ON tenant_user_cas(ca_pubkey);
-- exposures: one published guest port. The fleet binds host_port on the VM's
-- host and pipes it to guest_port inside the guest. host_id is denormalized
-- from the VM at create so allocation is a plain transaction against one
-- table, and the unique index IS the collision check — a port is claimed by
-- whoever inserts first, and the second insert fails at the storage layer
-- rather than after a read-then-write race.
CREATE TABLE IF NOT EXISTS exposures (
id TEXT PRIMARY KEY,
tenant TEXT NOT NULL REFERENCES tenants(id),
vm_id TEXT NOT NULL REFERENCES vms(id) ON DELETE CASCADE,
host_id TEXT NOT NULL REFERENCES hosts(id),
guest_port INTEGER NOT NULL,
host_port INTEGER NOT NULL,
protocol TEXT NOT NULL DEFAULT 'tcp',
scope TEXT NOT NULL DEFAULT 'lan',
created_at DATETIME NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS exposures_host_port_proto ON exposures(host_id, host_port, protocol);
-- Volumes. A claim is the tenant's request for storage; a volume is the
-- fleet's placement of bytes on one host, created by the first VM that names
-- the claim. "Unbound" is the absence of a volumes row, never a NULL.
CREATE TABLE IF NOT EXISTS volume_claims (
id TEXT PRIMARY KEY,
tenant TEXT NOT NULL REFERENCES tenants(id),
name TEXT NOT NULL,
size_gb INTEGER NOT NULL,
bound_volume_id TEXT,
created_at DATETIME NOT NULL,
deleted_at DATETIME
);
CREATE UNIQUE INDEX IF NOT EXISTS volume_claims_tenant_name ON volume_claims(tenant, name) WHERE deleted_at IS NULL;
CREATE TABLE IF NOT EXISTS volumes (
id TEXT PRIMARY KEY,
host_id TEXT NOT NULL REFERENCES hosts(id),
claim_id TEXT NOT NULL REFERENCES volume_claims(id),
size_gb INTEGER NOT NULL,
created_at DATETIME NOT NULL,
deleted_at DATETIME
);
-- The database refuses a double attach; nothing above it has to remember to.
-- Hard-deleting a reaped VM takes its attachment with it, like exposures.
CREATE TABLE IF NOT EXISTS volume_attachments (
claim_id TEXT NOT NULL REFERENCES volume_claims(id),
vm_id TEXT NOT NULL REFERENCES vms(id) ON DELETE CASCADE
);
CREATE UNIQUE INDEX IF NOT EXISTS volume_attachments_claim ON volume_attachments(claim_id);
-- Console sessions. Server-side so revocation works and restarts keep
-- users signed in. id holds the SHA-256 of the session value, never the value
-- itself (see CreateSession); expiry enforced on read.
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
tenant TEXT NOT NULL REFERENCES tenants(id),
created_at DATETIME NOT NULL,
expires_at DATETIME NOT NULL
);
-- Personal access tokens, tenant-scoped, stored as SHA-256 of the secret.
-- expires_at NULL means non-expiring.
CREATE TABLE IF NOT EXISTS api_tokens (
id TEXT PRIMARY KEY,
tenant TEXT NOT NULL REFERENCES tenants(id),
name TEXT NOT NULL,
token_hash TEXT NOT NULL UNIQUE,
created_at DATETIME NOT NULL,
expires_at DATETIME,
last_used_at DATETIME,
revoked_at DATETIME
);
-- append-only operational audit trail (enrollment, decommission). Read via
-- ListAudit / GET /api/v1/audit; rows are never UPDATEd, and the only DELETE
-- is retention pruning (PruneAudit, driven by the server's audit_retention
-- config). The at column is second-precision UTC RFC3339, which makes
-- lexicographic comparison chronological -- PruneAudit's DELETE relies on
-- every writer keeping that format.
--
-- The tenant column (added via ensureColumn in Open — the table shipped
-- without it) scopes audit read per tenant. Rows written before any tenant is
-- known — e.g. host.enroll.denied from an UNAUTHENTICATED enroll attempt — are
-- recorded against SystemTenant, since a row must exist before an authenticated
-- tenant is resolvable. SystemTenant is a scope no tenant-scoped read surfaces,
-- so these stay out of every tenant's audit view rather than landing in one.
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
at DATETIME NOT NULL,
action TEXT NOT NULL,
detail TEXT NOT NULL DEFAULT ''
);
INSERT INTO meta(key, value) VALUES ('epoch', '0') ON CONFLICT DO NOTHING;
INSERT INTO meta(key, value) VALUES ('next_cidr_index', '1') ON CONFLICT DO NOTHING;
`
func Open(path, cidrPool string) (*Store, error) {
dsn := path + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)"
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, fmt.Errorf("open db: %w", err)
}
db.SetMaxOpenConns(1)
if _, err := db.Exec(schema); err != nil {
db.Close()
return nil, fmt.Errorf("apply schema: %w", err)
}
// Columns added after their table shipped: OIDC identity binding on tenants
// and the tenant scope on audit_log. Idempotent; old databases pick them up.
for _, c := range []struct{ table, column, decl string }{
{"tenants", "oidc_issuer", "TEXT NOT NULL DEFAULT ''"},
{"tenants", "oidc_subject", "TEXT NOT NULL DEFAULT ''"},
{"tenants", "email", "TEXT NOT NULL DEFAULT ''"},
// The 'default' literals are backfill stamps for rows that predate the
// tenant column (all writes pass the tenant explicitly, so the DEFAULT
// never applies to new rows).
{"audit_log", "tenant", "TEXT NOT NULL DEFAULT 'default'"},
{"revoked_ssh_certs", "tenant", "TEXT NOT NULL DEFAULT 'default'"},
// What eitri put in the guest's authorized_keys, described rather than
// copied: type, SHA256 fingerprint and comment. Derived ONCE at create
// (see api.describeKey) because the SSE snapshot marshals every VM on a
// 1s tick and parsing a key per read would be work repeated forever.
// Rows that predate these columns show no key, which is honest — the
// record was never taken.
{"vms", "injected_key_type", "TEXT NOT NULL DEFAULT ''"},
{"vms", "injected_key_fp", "TEXT NOT NULL DEFAULT ''"},
{"vms", "injected_key_comment", "TEXT NOT NULL DEFAULT ''"},
// The public half of the guest's SSH host key, as its host reported it.
// The private half is the host's and stays there; this is the half the
// control plane signs a certificate for.
{"vms", "ssh_host_pubkey", "TEXT NOT NULL DEFAULT ''"},
// The tenant user-CA set this VM was created against, as JSON. NULLABLE
// on purpose, and the only vms column that is: every other late column
// backfills to a zero value that reads as "nothing to say", but here
// the empty set and the absent record mean opposite things — "this
// guest trusts no CA" versus "we did not write down which CAs it
// trusts". Rows that predate this column get NULL and say so.
{"vms", "trusted_cas", "TEXT"},
// Rollback ballast, and nothing else: no code in this release reads or
// writes this column, and no VM's behaviour depends on it. It is here
// because v0.0.5's Open runs `UPDATE vms SET persistent = 1` before it
// will serve anything, so a database this binary has touched must still
// carry the column or the previous release cannot start on it — and a
// release whose recovery plan is "roll back" would leave the plane down
// in both directions. Carried here rather than in the CREATE TABLE above
// so the guarantee covers a database this release CREATED as well as one
// it upgraded; a fresh plane has to be rollable too. NOT NULL DEFAULT 1
// because nothing writes it and every VM is persistent, so v0.0.5 reads
// back exactly the value it would have written itself.
//
// The real drop lands the release AFTER this one, alongside reserving
// proto field 9 — the same discipline for the same reason: retire the
// compatibility only once no v0.0.5 can still be out there.
{"vms", "persistent", "INTEGER NOT NULL DEFAULT 1"},
// The named host network this VM attaches to, as recorded at create.
// '' is the NAT underlay — every row that predates this column
// backfills to it, which is exactly what an unset network means.
{"vms", "network", "TEXT NOT NULL DEFAULT ''"},
// The address the site's DHCP server granted a guest on its named
// network, as its host snooped it. Agent-reported like assigned_ip and
// separate from it: assigned_ip is the private-fabric address every
// guest has from boot, this one exists only for a guest with a second
// NIC and only once that network has answered. '' is "not discovered",
// which is also what every row predating the column honestly says.
{"vms", "network_ip", "TEXT NOT NULL DEFAULT ''"},
// The address a host presents on the network it reaches the fleet
// over. Reported every tick like the guest subnet, and stored for the
// same reason: the console renders `host:port` for every exposure,
// including on a host that is momentarily offline.
{"hosts", "uplink_addr", "TEXT NOT NULL DEFAULT ''"},
} {
if err := ensureColumn(db, c.table, c.column, c.decl); err != nil {
db.Close()
return nil, err
}
}
// freed_cidrs held bridge CIDRs returned by a departing host, to be re-issued
// before the monotonic allocator was consulted. Nothing is recycled now: the
// column holds what the HOST claims, so re-issuing it would hand one host's
// subnet to another as a suggestion.
if err := dropTable(db, "freed_cidrs"); err != nil {
db.Close()
return nil, err
}
// Revocation became tenant-scoped; the old key would silently drop a second
// tenant's revocation of the same serial. See rekeyRevokedSSHCerts.
if err := rekeyRevokedSSHCerts(db); err != nil {
db.Close()
return nil, err
}
// A guest's host private key belongs to its host, so there is no column for
// one. A database written before that was true has both the column and the
// keys in it; dropping the column takes them with it. The certificate beside
// it is public, is what clients verify the guest by, and is untouched.
if err := dropColumn(db, "vms", "ssh_host_key"); err != nil {
db.Close()
return nil, err
}
// A host port is claimed per protocol, so TCP 30000 and UDP 30000 are two
// grants rather than a collision. The index that spanned only (host, port)
// would refuse the second one, so it goes — exposures_host_port_proto above
// has already replaced it by the time this runs.
if err := dropIndex(db, "exposures_host_port"); err != nil {
db.Close()
return nil, err
}
// One identity binds at most one tenant (per issuer). Partial index so
// unbound rows (empty issuer+subject) don't collide.
if _, err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS tenants_identity
ON tenants(oidc_issuer, oidc_subject) WHERE oidc_subject != ''`); err != nil {
db.Close()
return nil, fmt.Errorf("create tenants_identity index: %w", err)
}
// Store cidr_pool; ON CONFLICT DO NOTHING means the first call wins.
if _, err := db.Exec(`INSERT INTO meta(key, value) VALUES ('cidr_pool', ?) ON CONFLICT DO NOTHING`, cidrPool); err != nil {
db.Close()
return nil, fmt.Errorf("seed cidr_pool: %w", err)
}
var stored string
if err := db.QueryRow(`SELECT value FROM meta WHERE key='cidr_pool'`).Scan(&stored); err != nil {
db.Close()
return nil, fmt.Errorf("read cidr_pool: %w", err)
}
if stored != cidrPool {
slog.Warn("--cidr-pool was ignored: this database was created with a different guest-subnet pool, "+
"and hosts keep the subnet they were allocated, so the stored pool stands and every host "+
"enrolled from now on is still allocated out of it",
"stored", stored, "passed", cidrPool)
}
s := &Store{db: db, dbDir: filepath.Dir(path)}
// The last migration, and the only one that writes rows rather than shape:
// a VM that has no recorded CA set is given one now, so that a set frozen at
// create is the only answer anything serves. See backfillTrustedCAs.
if err := s.backfillTrustedCAs(); err != nil {
db.Close()
return nil, err
}
return s, nil
}
func (s *Store) Close() error { return s.db.Close() }
// Ping verifies the database handle is live with a cheap round-trip. Used by
// the readiness probe; the context bounds a wedged driver.
func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) }
// Epoch reads the current epoch value directly. Production callers get the
// epoch via SpecForHost (paired with a matching desired-VM read in the
// same tx); this exists as a test/observability hook for asserting exactly
// which mutations bump the epoch.
func (s *Store) Epoch() (uint64, error) {
var v uint64
err := s.db.QueryRow(`SELECT CAST(value AS INTEGER) FROM meta WHERE key='epoch'`).Scan(&v)
return v, err
}
func bumpEpoch(tx *sql.Tx) error {
_, err := tx.Exec(`UPDATE meta SET value = CAST(value AS INTEGER)+1 WHERE key='epoch'`)
return err
}
// subnetForIndex returns the idx-th /24 within pool using 32-bit arithmetic.
// idx starts at 1; the 0th /24 (the pool's own network address block) is reserved.
// Returns an error when the resulting /24 is outside the pool (exhaustion).
func subnetForIndex(pool netip.Prefix, idx int64) (string, error) {
if idx > 1<<23 {
return "", fmt.Errorf("cidr pool %s exhausted at host index %d", pool, idx)
}
base := pool.Masked().Addr().As4()
b := uint32(base[0])<<24 | uint32(base[1])<<16 | uint32(base[2])<<8 | uint32(base[3])
start := b + uint32(idx)*256 // idx-th /24; each /24 is 256 addresses
last := start + 255
startAddr := netip.AddrFrom4([4]byte{byte(start >> 24), byte(start >> 16), byte(start >> 8), byte(start)})
lastAddr := netip.AddrFrom4([4]byte{byte(last >> 24), byte(last >> 16), byte(last >> 8), byte(last)})
if !pool.Contains(startAddr) || !pool.Contains(lastAddr) {
return "", fmt.Errorf("cidr pool %s exhausted at host index %d", pool, idx)
}
return fmt.Sprintf("%d.%d.%d.0/24", start>>24&0xff, start>>16&0xff, start>>8&0xff), nil
}
// CreateEnrollmentToken mints a single-use enroll token bound to tenant. The
// enroll endpoint is unauthenticated, so the token IS the tenant credential —
// the enrolling host lands in this tenant. Tenant existence is checked here
// (the column carries no FK; the table is transient).
func (s *Store) CreateEnrollmentToken(tenant string) (string, error) {
var one int
if err := s.db.QueryRow(`SELECT 1 FROM tenants WHERE id=?`, tenant).Scan(&one); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return "", fmt.Errorf("unknown tenant %q", tenant)
}
return "", fmt.Errorf("check tenant: %w", err)
}
tok := random.Hex(32)
h := sha256.Sum256([]byte(tok))
hash := hex.EncodeToString(h[:])
expiresAt := time.Now().UTC().Add(15 * time.Minute)
_, err := s.db.Exec(
`INSERT INTO enrollment_tokens(token_hash, expires_at, tenant) VALUES (?, ?, ?)`,
hash, expiresAt.Format(time.RFC3339), tenant,
)
if err != nil {
return "", fmt.Errorf("insert token: %w", err)
}
return tok, nil
}
// EnrollFacts is what a joining host says about itself. It is a struct rather
// than more positional strings because the enrollment transaction already
// carried six of them, and because BridgeCIDR needs three states that a plain
// string cannot express — see its field.
type EnrollFacts struct {
Name, OS, Arch, Provisioner string
// Remote is the client address, recorded in the audit row.
Remote string
// BridgeCIDR is the subnet this host says its guests are on, and it is a
// POINTER because absent and empty mean different things. Nil is "this host
// has no opinion" — an older agent, or a Linux host started without
// --bridge-cidr — and takes the pool's suggestion. A non-nil empty string is
// "this host deliberately has none", which is what a Mac says: its OS owns
// the guest network and will report the subnet later, so allocating one for
// it would invent a fact. A non-nil value is stored as given.
BridgeCIDR *string
}
// RedeemEnrollmentToken atomically consumes tok and creates the host row. The
// created host inherits the tenant bound to the consumed token. remote (the
// enrolling client's IP) is recorded in a host.enroll audit row written in
// the SAME transaction, so an enrolled host can never exist without its
// durable audit record.
func (s *Store) RedeemEnrollmentToken(tok string, f EnrollFacts) (Host, error) {
h := sha256.Sum256([]byte(tok))
hash := hex.EncodeToString(h[:])
tx, err := s.db.Begin()
if err != nil {
return Host{}, err
}
defer tx.Rollback()
now := time.Now().UTC()
res, err := tx.Exec(
`UPDATE enrollment_tokens SET used_at=? WHERE token_hash=? AND used_at IS NULL AND expires_at > ?`,
now.Format(time.RFC3339), hash, now.Format(time.RFC3339),
)
if err != nil {
return Host{}, fmt.Errorf("mark token used: %w", err)
}
n, _ := res.RowsAffected()
if n != 1 {
return Host{}, fmt.Errorf("token invalid, expired, or already used")
}
var tenant string
if err := tx.QueryRow(`SELECT tenant FROM enrollment_tokens WHERE token_hash=?`, hash).Scan(&tenant); err != nil {
return Host{}, fmt.Errorf("read token tenant: %w", err)
}
var cidrPool string
var nextIdx int64
if err := tx.QueryRow(`SELECT value FROM meta WHERE key='cidr_pool'`).Scan(&cidrPool); err != nil {
return Host{}, fmt.Errorf("read cidr_pool: %w", err)
}
if err := tx.QueryRow(`SELECT CAST(value AS INTEGER) FROM meta WHERE key='next_cidr_index'`).Scan(&nextIdx); err != nil {
return Host{}, fmt.Errorf("read next_cidr_index: %w", err)
}
prefix, err := netip.ParsePrefix(cidrPool)
if err != nil {
return Host{}, fmt.Errorf("parse cidr_pool: %w", err)
}
// The host proposes; the fleet only suggests. A host that states its subnet
// is recorded as given — validated as a network, never checked against the
// pool, because a host whose OS owns the guest network is on a subnet no
// allocation of ours will ever contain.
//
// Nothing is recycled back into the pool. Post-inversion this column holds
// what the HOST claims, so returning it would hand a Mac's 192.168.64.0/24
// to the next Linux host as its suggested bridge. A /16 pool is 65k /24s and
// the monotonic index alone is sufficient: leaking indices is cheaper than
// re-issuing a value the fleet did not choose.
var bridgeCIDR string
switch {
case f.BridgeCIDR != nil && *f.BridgeCIDR != "":
if err := validGuestCIDR(*f.BridgeCIDR); err != nil {
return Host{}, err
}
bridgeCIDR = *f.BridgeCIDR
case f.BridgeCIDR != nil:
// Deliberately none: this host will report its subnet once it can see
// it. Allocating one here would invent a fact and put it on the row.
bridgeCIDR = ""
default:
// No opinion — an older agent, or a host that wants the fleet's advice.
bridgeCIDR, err = subnetForIndex(prefix, nextIdx)
if err != nil {
return Host{}, err
}
if _, err := tx.Exec(`UPDATE meta SET value=? WHERE key='next_cidr_index'`, nextIdx+1); err != nil {
return Host{}, fmt.Errorf("increment next_cidr_index: %w", err)
}
}
id := random.Hex(16)
if _, err := tx.Exec(
`INSERT INTO hosts(id, name, os, arch, provisioner, bridge_cidr, enrolled_at, tenant) VALUES (?,?,?,?,?,?,?,?)`,
id, f.Name, f.OS, f.Arch, f.Provisioner, bridgeCIDR, now.Format(time.RFC3339), tenant,
); err != nil {
return Host{}, fmt.Errorf("insert host: %w", err)
}
// Audit in the SAME tx: the durable record of who enrolled cannot be lost
// once the enrollment itself commits. The token appears only as a hash
// prefix (matches the enrollment_tokens.token_hash the mint row logs).
detail, _ := json.Marshal(map[string]string{
"host_id": id, "name": f.Name, "os": f.OS, "arch": f.Arch,
"remote": f.Remote, "token_hash_prefix": hash[:8],
})
if _, err := tx.Exec(`INSERT INTO audit_log(at, tenant, action, detail) VALUES (?, ?, ?, ?)`,
now.Format(time.RFC3339), tenant, "host.enroll", string(detail)); err != nil {
return Host{}, fmt.Errorf("audit enroll: %w", err)
}
if err := tx.Commit(); err != nil {
return Host{}, err
}
return Host{
ID: id,
Name: f.Name,
OS: f.OS,
Arch: f.Arch,
Provisioner: f.Provisioner,
BridgeCIDR: bridgeCIDR,
Status: "enrolled",
CredGeneration: 1,
EnrolledAt: now,
Tenant: tenant,
}, nil
}
// hostColumns is the positional column list every host SELECT uses, so the
// order stays locked to the positional Scans in GetHost and listHosts below.
// Adding a column is one edit here plus the two Scans — no query can drift out
// of lockstep. Mirrors vmColumns/exposureColumns/tenantColumns.
const hostColumns = `id, name, os, arch, provisioner, bridge_cidr, status, enrolled_at, cred_generation, tenant, os_id, os_pretty, os_version, kernel, cpu_model, virt, uplink_addr`
func (s *Store) GetHost(id string) (Host, error) {
var h Host
var enrolledAt string
err := s.db.QueryRow(
`SELECT `+hostColumns+` FROM hosts WHERE id=?`, id,
).Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt, &h.UplinkAddr)
if err != nil {
return Host{}, err
}
h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)
return h, nil
}
// validGuestCIDR is the sanity check applied to every subnet a host claims,
// wherever it arrives — proposed at enrollment, or reported in a tick. It is
// the same check netenv makes before handing the value to the kernel, applied
// on the receiving side so a garbage string cannot reach the console, an
// operator's eyes, or anything that reads the row.
//
// It asks whether the value is a network. It never asks WHICH network: a host's
// guest subnet is that host's business, and applying sanity to one of a row's
// two network columns and not the other would be sanity by coincidence.
func validGuestCIDR(cidr string) error {
p, err := netip.ParsePrefix(cidr)
if err != nil {
return fmt.Errorf("guest cidr %q: %w", cidr, err)
}
if !p.Addr().Is4() {
return fmt.Errorf("guest cidr %q must be IPv4", cidr)
}
return nil
}
// RecordHostNetwork stores the subnet a host says its guests are on. The value
// is free text supplied by whoever claims to be that host, so it is validated
// the same way netenv validates it before handing it to the kernel — parses as
// a prefix, and IPv4 — applied on the receiving side so a garbage string cannot
// reach the console, an operator's eyes, or anything that reads the row.
//
// This is sanity, not topology: it asks whether the value is a network, never
// which network it ought to be. A host's guest subnet is that host's business.
//
// The WHERE clause makes a concurrent reconnect harmless and a repeat free at
// the storage layer; the caller still guards the round trip, because reports
// arrive every tick forever and the store runs on one connection.
func (s *Store) RecordHostNetwork(id, cidr string) error {
if err := validGuestCIDR(cidr); err != nil {
return err
}
_, err := s.db.Exec(`UPDATE hosts SET bridge_cidr=? WHERE id=? AND bridge_cidr<>?`, cidr, id, cidr)
return err
}
// RecordHostUplink stores the address a host says it presents on the network it
// reaches the fleet over. The value is free text supplied by whoever claims to
// be that host, so it is checked the way every reported network fact is: it
// must parse as an address. It asks whether the value IS an address, never
// which address it ought to be — a host's own uplink is that host's business.
//
// An address that reaches nobody — loopback, unspecified, link-local,
// multicast — is refused rather than stored: an operator dials this value, so a
// row holding somewhere unreachable is worse than a row holding nothing. The
// caller keeps what it already knows.
//
// The WHERE clause makes a repeat free at the storage layer; the caller still
// guards the round trip, because reports arrive every tick forever.
func (s *Store) RecordHostUplink(id, addr string) error {
if !usableAddress(addr) {
return fmt.Errorf("host uplink %q: not an address anyone can reach", addr)
}
_, err := s.db.Exec(`UPDATE hosts SET uplink_addr=? WHERE id=? AND uplink_addr<>?`, addr, id, addr)
return err
}
// UpdateHostFacts refreshes what a host reports about itself. Every column
// keeps its prior value when the reported one is EMPTY, because silence is not
// a statement: an agent too old to send a field, or a Hello carrying no facts
// at all, would otherwise erase what an earlier one told us. A host with
// genuinely nothing to say sends a value saying so — a Mac reports virt "none",
// not "".
func (s *Store) UpdateHostFacts(id string, f HostFacts) error {
_, err := s.db.Exec(
`UPDATE hosts SET
os_id = CASE WHEN ?='' THEN os_id ELSE ? END,
os_pretty = CASE WHEN ?='' THEN os_pretty ELSE ? END,
os_version = CASE WHEN ?='' THEN os_version ELSE ? END,
kernel = CASE WHEN ?='' THEN kernel ELSE ? END,
cpu_model = CASE WHEN ?='' THEN cpu_model ELSE ? END,
virt = CASE WHEN ?='' THEN virt ELSE ? END,
provisioner = CASE WHEN ?='' THEN provisioner ELSE ? END
WHERE id=?`,
f.OSID, f.OSID, f.OSPretty, f.OSPretty, f.OSVersion, f.OSVersion,
f.Kernel, f.Kernel, f.CPUModel, f.CPUModel, f.Virt, f.Virt,
f.Provisioner, f.Provisioner, id)
return err
}
// BumpCredGeneration increments the host's credential generation, revoking
// every credential minted at the previous generation, and returns the new
// value. The host.credential.revoke audit row is written in the SAME
// transaction — a security action must not be able to happen unrecorded.
// Errors (sql.ErrNoRows via %w) if the host does not exist.
func (s *Store) BumpCredGeneration(id, remote string) (int64, error) {
tx, err := s.db.Begin()
if err != nil {
return 0, err
}
defer tx.Rollback()
var gen int64
var tenant string
if err := tx.QueryRow(
`UPDATE hosts SET cred_generation = cred_generation + 1 WHERE id=? RETURNING cred_generation, tenant`, id,
).Scan(&gen, &tenant); err != nil {
return 0, fmt.Errorf("bump cred_generation for %s: %w", id, err)
}
detail, _ := json.Marshal(map[string]string{
"host_id": id, "new_generation": strconv.FormatInt(gen, 10), "remote": remote,
})
if _, err := tx.Exec(`INSERT INTO audit_log(at, tenant, action, detail) VALUES (?, ?, ?, ?)`,
time.Now().UTC().Format(time.RFC3339), tenant, "host.credential.revoke", string(detail)); err != nil {
return 0, fmt.Errorf("audit revoke: %w", err)
}
return gen, tx.Commit()
}
// querier is the subset of *sql.DB / *sql.Tx the list helpers need, so the
// same scan logic serves both the standalone reads and Snapshot's single-tx read.
type querier interface {
Query(query string, args ...any) (*sql.Rows, error)
}
func (s *Store) ListHosts() ([]Host, error) { return listHosts(s.db) }
func listHosts(q querier) ([]Host, error) {
rows, err := q.Query(`SELECT ` + hostColumns + ` FROM hosts`)
if err != nil {
return nil, err
}
defer rows.Close()
var hosts []Host
for rows.Next() {
var h Host
var enrolledAt string
if err := rows.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt, &h.UplinkAddr); err != nil {
return nil, err
}
h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)
hosts = append(hosts, h)
}
return hosts, rows.Err()
}
func (s *Store) CreateVM(vm VM) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
now := time.Now().UTC()
if vm.ID == "" {
vm.ID = random.Hex(16)
}
var hostStatus, hostTenant string
switch err := tx.QueryRow(`SELECT status, tenant FROM hosts WHERE id=?`, vm.HostID).Scan(&hostStatus, &hostTenant); {
case errors.Is(err, sql.ErrNoRows):
return ErrHostNotFound
case err != nil:
return fmt.Errorf("lookup host status: %w", err)
case hostStatus != "enrolled":
return ErrHostNotEnrolled
}
vm.Tenant = hostTenant
// The CA set is frozen here, in the same statement that makes the VM exist,
// so there is no window in which a row exists without the trust it was
// created against. A caller that supplies none writes NULL rather than an
// empty array — see VM.TrustedCAs for why those are different facts.
var trustedCAs any
if vm.TrustedCAs != nil {
b, err := json.Marshal(vm.TrustedCAs)
if err != nil {
return fmt.Errorf("marshal trusted cas: %w", err)
}
trustedCAs = string(b)
}
_, err = tx.Exec(
`INSERT INTO vms(id, host_id, name, tenant, image_url, image_sha256, cloud_init, ssh_authorized_key,
injected_key_type, injected_key_fp, injected_key_comment, trusted_cas,
vcpus, mem_mb, disk_gb, power_state, network, created_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
vm.ID, vm.HostID, vm.Name, vm.Tenant, vm.ImageURL, vm.ImageSHA256,
vm.CloudInit, vm.SSHAuthorizedKey,
vm.InjectedKeyType, vm.InjectedKeyFP, vm.InjectedKeyComment, trustedCAs,
vm.VCPUs, vm.MemMB, vm.DiskGB, vm.PowerState, vm.Network,
now.Format(time.RFC3339),
)
if err != nil {
// SQLITE_CONSTRAINT_UNIQUE (2067): the only UNIQUE constraint an
// insert can trip besides the PK (which is 1555 and random-hex) is
// vms_tenant_name — a name collision within the tenant. Matched by
// errno, not message text: the message embeds the index's column list
// and silently breaks on the next index change (review finding).
if serr, ok := errors.AsType[*sqlite.Error](err); ok {
switch serr.Code() {
case 2067: // SQLITE_CONSTRAINT_UNIQUE
return ErrNameTaken
case 787: // SQLITE_CONSTRAINT_FOREIGNKEY: tenant is derived in-tx
// from the host row above, and hosts.tenant itself carries an
// enforced FK, so vm.Tenant always references an existing
// tenant — it can't be what trips this. host_id -> hosts is
// the only FK left that a CreateVM insert can violate, so 787
// unambiguously means the host is gone.
return ErrHostNotFound
}
}
return fmt.Errorf("insert vm: %w", err)
}
// Claims bind HERE, in the same tx that places the VM: a claim the fleet
// cannot honour (another tenant's, already held, or already living on
// another host) takes the whole create down with it, so a refused VM
// leaves no row and no half-bound claim behind.
if err := bindClaims(tx, vm); err != nil {
return err
}
if err := bumpEpoch(tx); err != nil {
return fmt.Errorf("bump epoch: %w", err)
}
return tx.Commit()
}
// mutate runs a single mutation SQL that should affect exactly 1 row, then bumps the epoch, all in a tx.
func (s *Store) mutate(query string, args ...any) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
res, err := tx.Exec(query, args...)
if err != nil {
return err
}
n, _ := res.RowsAffected()
if n == 0 {
return sql.ErrNoRows
}
if err := bumpEpoch(tx); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) SetVMPower(id, power string) error {
return s.mutate(`UPDATE vms SET power_state=? WHERE id=? AND deleted_at IS NULL`, power, id)
}
func (s *Store) TombstoneVM(id string) error {
return s.mutate(`UPDATE vms SET deleted_at=? WHERE id=? AND deleted_at IS NULL`,
time.Now().UTC().Format(time.RFC3339), id)
}
// RestoreVM un-tombstones a VM that is still within the teardown grace window
// (row present, not yet hard-deleted): clears deleted_at so the agent re-adopts
// it. Returns sql.ErrNoRows if the row is not restorable — never deleted, or
// already reaped (the row is gone). Bumps the epoch so agents re-snapshot and
// re-add it to desired.
func (s *Store) RestoreVM(id string) error {
return s.mutate(`UPDATE vms SET deleted_at=NULL WHERE id=? AND deleted_at IS NOT NULL`, id)
}
// HardDeleteVM removes a tombstoned VM row on the ack from the host that holds
// it, and bumps the epoch. Only tombstoned rows are deletable — a live row gets
// sql.ErrNoRows, so an ack can never skip the teardown grace and leave a running
// guest orphaned.
//
// hostID is a predicate rather than a check made before the statement, the same
// idiom RecordVMNetworkIP and RecordVMHostKey use and for the same reason: a
// host may only ever speak for the VMs it holds, and a predicate leaves no
// read-then-write window in which the VM could move. The window this closes is
// the teardown grace itself: an ack naming a VM some other host holds would end
// that VM's restore window before its own host had reaped it. sql.ErrNoRows
// covers all three refusals — the row is gone, it is live, or another host
// holds it — and the caller cannot tell which.
func (s *Store) HardDeleteVM(id, hostID string) error {
return s.mutate(`DELETE FROM vms WHERE id=? AND host_id=? AND deleted_at IS NOT NULL`, id, hostID)
}
// ForceDeleteVM removes a tombstoned VM row on the server's own authority, with
// no host to name: the backstop for a tombstone whose host is gone or offline
// past its grace, where no agent will ever ack. Callers are the server's own
// sweeps, which read the row first and answer to their own authorization; the
// host-supplied path is HardDeleteVM, which is scoped.
func (s *Store) ForceDeleteVM(id string) error {
return s.mutate(`DELETE FROM vms WHERE id=? AND deleted_at IS NOT NULL`, id)
}
// DecommissionHost marks a host as decommissioning and tombstones all its live
// VMs so the agent reaps them through the normal quarantine→destroy path. One
// epoch bump for the whole transition.
func (s *Store) DecommissionHost(id string) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
res, err := tx.Exec(`UPDATE hosts SET status='decommissioning' WHERE id=?`, id)
if err != nil {
return fmt.Errorf("set host status: %w", err)
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows
}
if _, err := tx.Exec(
`UPDATE vms SET deleted_at=? WHERE host_id=? AND deleted_at IS NULL`,
time.Now().UTC().Format(time.RFC3339), id,
); err != nil {
return fmt.Errorf("tombstone host vms: %w", err)
}
if err := bumpEpoch(tx); err != nil {
return fmt.Errorf("bump epoch: %w", err)
}
return tx.Commit()
}
// Alloc is the sum of resources committed to live VMs on a host.
type Alloc struct{ VCPUs, MemMB, DiskGB int64 }
// allocatedByHost returns, per host, the resources allocated to its live
// (non-tombstoned) VMs. Hosts with no live VMs are absent from the map. Used
// by Snapshot; there is no standalone exported accessor (nothing outside the
// package needs allocation without hosts/VMs, and Snapshot is the consistent
// way to get all three together).
func allocatedByHost(q querier) (map[string]Alloc, error) {
rows, err := q.Query(`
SELECT host_id, COALESCE(SUM(vcpus),0), COALESCE(SUM(mem_mb),0), COALESCE(SUM(disk_gb),0)
FROM vms WHERE deleted_at IS NULL GROUP BY host_id`)
if err != nil {
return nil, err
}
defer rows.Close()
out := make(map[string]Alloc)
for rows.Next() {
var id string
var a Alloc
if err := rows.Scan(&id, &a.VCPUs, &a.MemMB, &a.DiskGB); err != nil {
return nil, err
}
out[id] = a
}
return out, rows.Err()
}
// Commitment is everything a host is holding, split by why it is holding it.
// Live is the VMs it is meant to be running; Pending is the tombstoned rows —
// deletes it has been asked for and not yet finished — and PendingVMs is how
// many rows those are. Held() is the sum, which is the number a placement has
// to beat; the split exists so a refusal can say which half is binding, because
// the two have different remedies (delete something, or wait).
type Commitment struct {
Live Alloc
Pending Alloc
PendingVMs int
}
// Held is what the host is holding in total: live VMs plus destroys in flight.
func (c Commitment) Held() Alloc {
return Alloc{
VCPUs: c.Live.VCPUs + c.Pending.VCPUs,
MemMB: c.Live.MemMB + c.Pending.MemMB,
DiskGB: c.Live.DiskGB + c.Pending.DiskGB,
}
}
// CommittedOnHost sums the specs of every VM row still on a host — what the
// machine is holding, which is the number a placement decision has to beat.
//
// It deliberately does NOT filter deleted_at, and that is the whole point of
// having it beside allocatedByHost. A tombstone is a destroy in progress, not a
// destroy: the row is marked, the agent is asked to tear the guest down, and
// only when it acks (or the abandoned sweep gives up on an offline host) is the
// row hard-deleted. Until that reap the disk is still on the host and the guest
// may still be shutting down, so its resources are not free to promise to
// somebody else. Reaped VMs need no predicate — their rows are gone.
//
// The two halves come back separately because counting a tombstone is right and
// blaming the caller for it is not: a create refused by resources a delete has
// not finished releasing has a different answer ("wait") than one refused by
// VMs that are actually there ("delete something"), and only the split can tell
// them apart.
//
// allocatedByHost, which feeds the console's "allocated", counts live rows
// only, so during a teardown Held() reads higher. Erring that way is the safe
// direction for admission: refusing a VM for a bed that is being stripped costs
// a retry, admitting one into it costs a failed VM.
//
// Live volumes on the host count into Live.DiskGB too. A bound volume is disk
// the host has committed WITH OR WITHOUT A VM: the bytes outlive every guest
// that mounts them, which is the point of a volume, so the space stays spoken
// for between one VM's destroy and the next VM's create.
func (s *Store) CommittedOnHost(hostID string) (Commitment, error) {
var c Commitment
err := s.db.QueryRow(`
SELECT COALESCE(SUM(CASE WHEN deleted_at IS NULL THEN vcpus END),0),
COALESCE(SUM(CASE WHEN deleted_at IS NULL THEN mem_mb END),0),
COALESCE(SUM(CASE WHEN deleted_at IS NULL THEN disk_gb END),0),
COALESCE(SUM(CASE WHEN deleted_at IS NOT NULL THEN vcpus END),0),
COALESCE(SUM(CASE WHEN deleted_at IS NOT NULL THEN mem_mb END),0),
COALESCE(SUM(CASE WHEN deleted_at IS NOT NULL THEN disk_gb END),0),
COUNT(deleted_at)
FROM vms WHERE host_id=?`, hostID).Scan(
&c.Live.VCPUs, &c.Live.MemMB, &c.Live.DiskGB,
&c.Pending.VCPUs, &c.Pending.MemMB, &c.Pending.DiskGB,
&c.PendingVMs)
if err != nil {
return c, err
}
// Only live volumes. A tombstoned one is a reclaim in flight and its file
// may still be on the disk, exactly like a tombstoned VM's — the honest
// place for it is Pending, and it is not counted at all until something
// asks for that distinction.
var vol int64
if err := s.db.QueryRow(
`SELECT COALESCE(SUM(size_gb),0) FROM volumes WHERE host_id=? AND deleted_at IS NULL`, hostID,
).Scan(&vol); err != nil {
return c, err
}
c.Live.DiskGB += vol
return c, nil
}
// AuditEntry is one row of the append-only audit trail.
type AuditEntry struct {
At time.Time
Action string
Detail string
}
// AppendAudit records an audit event scoped to tenant. detail is a small JSON
// blob; keep secrets out (hash prefixes, not tokens). Rows written before an
// authenticated tenant is known (e.g. a denied enroll attempt) pass
// SystemTenant, which no tenant-scoped read ever surfaces.
func (s *Store) AppendAudit(tenant, action, detail string) error {
_, err := s.db.Exec(`INSERT INTO audit_log(at, tenant, action, detail) VALUES (?, ?, ?, ?)`,
time.Now().UTC().Format(time.RFC3339), tenant, action, detail)
return err
}
// scanAuditRows drains an `at, action, detail`-shaped result set into
// AuditEntry values, newest-first per the caller's ORDER BY. Both audit queries
// share the same three columns and RFC3339 at-parse, so they share this.
func scanAuditRows(rows *sql.Rows) ([]AuditEntry, error) {
defer rows.Close()
var out []AuditEntry
for rows.Next() {
var e AuditEntry
var at string
if err := rows.Scan(&at, &e.Action, &e.Detail); err != nil {
return nil, err
}
e.At, _ = time.Parse(time.RFC3339, at)
out = append(out, e)
}
return out, rows.Err()
}
// ListAudit returns up to limit of tenant's audit entries, newest first.
func (s *Store) ListAudit(tenant string, limit int) ([]AuditEntry, error) {
rows, err := s.db.Query(`SELECT at, action, detail FROM audit_log WHERE tenant=? ORDER BY id DESC LIMIT ?`, tenant, limit)
if err != nil {
return nil, err
}
return scanAuditRows(rows)
}
// ListVMEvents returns up to limit of tenant's audit rows whose detail JSON
// carries the given vm_id (the lifecycle timeline for one VM), newest first. It
// filters on json_extract(detail,'$.vm_id'), so every lifecycle emitter must
// key the VM id as exactly "vm_id". Scoped to tenant so one tenant cannot read
// another's VM history. Historical events for a hard-deleted VM stay returnable:
// the append-only log outlives the VM row, and the vm.reap row carries the VM's
// tenant.
func (s *Store) ListVMEvents(tenant, vmID string, limit int) ([]AuditEntry, error) {
rows, err := s.db.Query(
`SELECT at, action, detail FROM audit_log WHERE tenant=? AND json_extract(detail,'$.vm_id')=? ORDER BY id DESC LIMIT ?`,
tenant, vmID, limit,
)
if err != nil {
return nil, err
}
return scanAuditRows(rows)
}
// PruneAudit deletes audit rows older than olderThan and reports how many
// were removed. Retention keeps the append-only log bounded; the caller
// (eitri-server) runs it at startup and daily.
func (s *Store) PruneAudit(olderThan time.Duration) (int64, error) {
if olderThan <= 0 {
return 0, nil
}
cutoff := time.Now().UTC().Add(-olderThan).Format(time.RFC3339)
res, err := s.db.Exec(`DELETE FROM audit_log WHERE at < ?`, cutoff)
if err != nil {
return 0, err
}
return res.RowsAffected()
}
// RevokedCert is one row of the SSH user-cert revocation list.
type RevokedCert struct {
Serial uint64
RevokedAt time.Time
Reason string
}
// RevokeSSHCert adds serial to tenant's revocation list so the gate rejects any
// cert carrying it. Idempotent: revoking an already-revoked serial is a no-op
// that keeps the ORIGINAL tenant/revoked_at/reason (a re-revoke does not
// overwrite the audit-relevant first record — including its owning tenant, so a
// second tenant cannot re-file a serial another already owns). serial is bit-cast
// to int64 for storage — SQLite INTEGER is signed 64-bit, and the cast is a
// bijection so uniqueness and lookups by serial are preserved.
func (s *Store) RevokeSSHCert(tenant string, serial uint64, reason string) error {
_, err := s.db.Exec(
`INSERT INTO revoked_ssh_certs(serial, tenant, revoked_at, reason) VALUES (?, ?, ?, ?)
ON CONFLICT(tenant, serial) DO NOTHING`,
int64(serial), tenant, time.Now().UTC().Format(time.RFC3339), reason,
)
if err != nil {
return fmt.Errorf("revoke ssh cert: %w", err)
}
return nil
}
// IsSSHCertRevoked reports whether tenant has revoked serial. The gate consults
// this on every cert authentication, so it is a hot read; the composite PRIMARY
// KEY makes it a single index lookup.
//
// Scoped on purpose: a certificate is revoked by the tenant whose CA signed it,
// and only for that tenant. Answering fleet-wide would let any tenant deny a
// serial belonging to another — the API cannot attribute a bare serial, because
// CAs are BYO and eitri never sees the certificates they mint.
func (s *Store) IsSSHCertRevoked(tenant string, serial uint64) (bool, error) {
var n int
err := s.db.QueryRow(`SELECT COUNT(*) FROM revoked_ssh_certs WHERE tenant=? AND serial=?`,
tenant, int64(serial)).Scan(&n)
if err != nil {
return false, fmt.Errorf("lookup revoked ssh cert: %w", err)
}
return n > 0, nil
}
// ListRevokedSSHCerts returns tenant's revoked cert serials (+ reason/time),
// newest first, for the tenant-scoped list endpoint.
func (s *Store) ListRevokedSSHCerts(tenant string) ([]RevokedCert, error) {
rows, err := s.db.Query(`SELECT serial, revoked_at, reason FROM revoked_ssh_certs WHERE tenant=? ORDER BY revoked_at DESC, serial DESC`, tenant)
if err != nil {
return nil, err
}
defer rows.Close()
var out []RevokedCert
for rows.Next() {
var rc RevokedCert
var serial int64
var revokedAt string
if err := rows.Scan(&serial, &revokedAt, &rc.Reason); err != nil {
return nil, err
}
rc.Serial = uint64(serial) // reverse the int64 bit-cast used at insert
rc.RevokedAt, _ = time.Parse(time.RFC3339, revokedAt)
out = append(out, rc)
}
return out, rows.Err()
}
// TenantUserCA is one registered per-tenant user-CA public key.
type TenantUserCA struct {
Tenant string
Pubkey string // canonical authorized_keys line (sshca.AuthorizedKeyLine)
Scope string
Label string
AddedBy string
CreatedAt time.Time
}
// AddTenantUserCA registers a user-CA pubkey for tenant. Idempotent: re-adding
// the same (tenant, ca_pubkey) keeps the original row (ON CONFLICT DO NOTHING).
func (s *Store) AddTenantUserCA(tenant, pubkey, scope, label, addedBy string) error {
if scope == "" {
scope = "tenant"
}
_, err := s.db.Exec(
`INSERT INTO tenant_user_cas(tenant, ca_pubkey, scope, label, added_by, created_at)
VALUES (?,?,?,?,?,?) ON CONFLICT(tenant, ca_pubkey) DO NOTHING`,
tenant, pubkey, scope, label, addedBy, time.Now().UTC().Format(time.RFC3339),
)
if err != nil {
return fmt.Errorf("add tenant user ca: %w", err)
}
return nil
}
// ListTenantUserCAs returns tenant's registered user CAs, newest first.
func (s *Store) ListTenantUserCAs(tenant string) ([]TenantUserCA, error) {
rows, err := s.db.Query(
`SELECT tenant, ca_pubkey, scope, label, added_by, created_at
FROM tenant_user_cas WHERE tenant=? ORDER BY created_at DESC, ca_pubkey`, tenant)
if err != nil {
return nil, err
}
defer rows.Close()
var out []TenantUserCA
for rows.Next() {
var c TenantUserCA
var created string
if err := rows.Scan(&c.Tenant, &c.Pubkey, &c.Scope, &c.Label, &c.AddedBy, &created); err != nil {
return nil, err
}
c.CreatedAt, _ = time.Parse(time.RFC3339, created)
out = append(out, c)
}
return out, rows.Err()
}
// RemoveTenantUserCA deletes a registered CA. Removing an absent one is a no-op.
func (s *Store) RemoveTenantUserCA(tenant, pubkey string) error {
_, err := s.db.Exec(`DELETE FROM tenant_user_cas WHERE tenant=? AND ca_pubkey=?`, tenant, pubkey)
if err != nil {
return fmt.Errorf("remove tenant user ca: %w", err)
}
return nil
}
// TenantHasUserCA reports whether tenant has ≥1 registered user CA. Gates VM
// create — a VM with no trusted CA would be unreachable.
func (s *Store) TenantHasUserCA(tenant string) (bool, error) {
var n int
if err := s.db.QueryRow(`SELECT COUNT(*) FROM tenant_user_cas WHERE tenant=?`, tenant).Scan(&n); err != nil {
return false, fmt.Errorf("count tenant user cas: %w", err)
}
return n > 0, nil
}
// TenantForUserCA returns the tenant that registered pubkey (canonical line).
// The gate calls this per cert auth to resolve the connection's tenant from the
// cert's signature key. ok=false ⇒ unregistered CA ⇒ reject.
func (s *Store) TenantForUserCA(pubkey string) (tenant string, ok bool, err error) {
err = s.db.QueryRow(`SELECT tenant FROM tenant_user_cas WHERE ca_pubkey=?`, pubkey).Scan(&tenant)
if err == sql.ErrNoRows {
return "", false, nil
}
if err != nil {
return "", false, fmt.Errorf("tenant for user ca: %w", err)
}
return tenant, true, nil
}
// RemoveHost finalizes decommission: it deletes the host row, refusing
// (in-transaction) while any VM rows remain for the host (not yet reaped), and
// while any volume rows do — live or tombstoned, because either way the bytes
// are still on that disk.
// Nothing is recycled — next_cidr_index only moves forward, so a removed host's
// subnet is retired with it rather than handed to the next host to enroll.
//
// A held volume is a different refusal from an undrained VM and returns
// ErrHostHoldsVolumes to say so. A VM drains on its own; a volume never does,
// because outliving its guests is the whole point of one. The operator has to
// delete the claims, let the agent reclaim them, and the sweep then succeeds.
func (s *Store) RemoveHost(id string) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
var n, vols int
if err := tx.QueryRow(`SELECT
(SELECT COUNT(*) FROM vms WHERE host_id=?),
(SELECT COUNT(*) FROM volumes WHERE host_id=?)`, id, id).Scan(&n, &vols); err != nil {
return fmt.Errorf("count host vms: %w", err)
}
if n > 0 {
return fmt.Errorf("host %s still has %d VM(s); not drained", id, n)
}
if vols > 0 {
return fmt.Errorf("host %s still has %d volume(s); delete their claims first: %w", id, vols, ErrHostHoldsVolumes)
}
res, err := tx.Exec(`DELETE FROM hosts WHERE id=?`, id)
if err != nil {
return fmt.Errorf("delete host: %w", err)
}
if rows, _ := res.RowsAffected(); rows == 0 {
return sql.ErrNoRows
}
if err := bumpEpoch(tx); err != nil {
return fmt.Errorf("bump epoch: %w", err)
}
return tx.Commit()
}
// ForceRemoval is the tally a forced host removal returns: what it destroyed
// and what it handed back. The one path that loses data on purpose has to be
// able to say what it lost, so the audit row can say it too — "the host is
// gone" is not an answer to "which volumes went with it".
//
// VolumeIDs names the destroyed volumes, in the order the store lists them
// (by id), so the same removal reports the same way twice. VMsPurged counts VM
// rows, live and tombstoned alike. ClaimsUnbound counts only the claims that
// LIVE ON as Pending: a claim already tombstoned here is unbound and then
// deleted outright, and nothing is waiting for it afterwards.
type ForceRemoval struct {
VMsPurged int
VolumesDestroyed int
ClaimsUnbound int
VolumeIDs []string
}
// ForceRemoveHost finalizes a host whose agent will never drain it (dead
// hardware): it purges every VM row for the host and deletes the host row, both
// in one transaction. Like RemoveHost it recycles nothing. Unlike the
// graceful path it does NOT wait for the agent to ack destroys, so it must only
// be used when the host is known gone; any VMs still physically running are
// orphaned with the hardware. Returns the tally of what went.
//
// THIS IS THE ONE PATH THAT LOSES DATA ON PURPOSE. The host's volumes go with
// the host, because that is where the bytes were: a live claim placed here
// returns to Pending and the next VM that names it places it somewhere else,
// with nothing of the old contents. A claim already tombstoned here is deleted
// outright — the reclaim it was waiting for can never be acked by a host that
// is gone, so keeping the row would only leave an unreapable tombstone behind.
// The operator asking for force has already decided the hardware is lost;
// refusing to admit its disks went with it would help nobody.
func (s *Store) ForceRemoveHost(id string) (ForceRemoval, error) {
var out ForceRemoval
tx, err := s.db.Begin()
if err != nil {
return out, err
}
defer tx.Rollback()
// Existence check only: nothing is reclaimed from a departing host, so the
// row's contents no longer matter here.
var exists int
switch err := tx.QueryRow(`SELECT 1 FROM hosts WHERE id=?`, id).Scan(&exists); {
case errors.Is(err, sql.ErrNoRows):
return out, sql.ErrNoRows
case err != nil:
return out, fmt.Errorf("lookup host: %w", err)
}
// The volumes whose bytes lived on this host and the claims they were
// placed for, read BEFORE anything is deleted and drained to completion
// before the first write — the store runs one connection, so an open
// result set blocks every statement below.
volumeIDs, claimIDs, err := volumesOnHost(tx, id)
if err != nil {
return out, err
}
// Unbind first, while the volumes rows this matches still exist.
res, err := tx.Exec(`UPDATE volume_claims SET bound_volume_id=NULL
WHERE bound_volume_id IN (SELECT id FROM volumes WHERE host_id=?)`, id)
if err != nil {
return out, fmt.Errorf("unbind host claims: %w", err)
}
unbound, _ := res.RowsAffected()
// VMs before volumes: deleting a VM cascades its attachments away, and a
// volume_claims row cannot go while an attachment still references it.
res, err = tx.Exec(`DELETE FROM vms WHERE host_id=?`, id)
if err != nil {
return out, fmt.Errorf("purge host vms: %w", err)
}
purged, _ := res.RowsAffected()
res, err = tx.Exec(`DELETE FROM volumes WHERE host_id=?`, id)
if err != nil {
return out, fmt.Errorf("purge host volumes: %w", err)
}
destroyed, _ := res.RowsAffected()
// A claim already tombstoned here has nothing left to reclaim: its host is
// gone and no ack is coming. Live claims survive as Pending, so a claim
// deleted here is one the unbind count above must not claim to have freed.
var tombstoned int64
for _, claimID := range claimIDs {
res, err := tx.Exec(`DELETE FROM volume_claims WHERE id=? AND deleted_at IS NOT NULL`, claimID)
if err != nil {
return out, fmt.Errorf("purge tombstoned claim: %w", err)
}
n, _ := res.RowsAffected()
tombstoned += n
}
if _, err := tx.Exec(`DELETE FROM hosts WHERE id=?`, id); err != nil {
return out, fmt.Errorf("delete host: %w", err)
}
if err := bumpEpoch(tx); err != nil {
return out, fmt.Errorf("bump epoch: %w", err)
}
if err := tx.Commit(); err != nil {
return out, err
}
return ForceRemoval{
VMsPurged: int(purged),
VolumesDestroyed: int(destroyed),
ClaimsUnbound: int(unbound - tombstoned),
VolumeIDs: volumeIDs,
}, nil
}
// usableAddress reports whether ip is an address something could actually be
// reached on — a guest on its host's network, or a host on the network it
// reaches the fleet over. It asks nothing about topology: a host's guests live
// wherever that host's network puts them — inside a Linux bridge the fleet
// allocated, or on whatever subnet macOS's vmnet happens to run — and the fleet
// is told that subnet, it does not decide it. So the only thing worth rejecting
// here is a value that names nobody under any topology: unparseable, the
// unspecified address, loopback, link-local (the 169.254/16 a guest reports when
// DHCP never answered), or multicast.
func usableAddress(ip string) bool {
addr, err := netip.ParseAddr(ip)
if err != nil {
return false
}
return !addr.IsUnspecified() && !addr.IsLoopback() &&
!addr.IsLinkLocalUnicast() && !addr.IsLinkLocalMulticast() && !addr.IsMulticast()
}
// RecordVMStatus persists one VM's (status, last_error, assigned_ip) as reported
// by the host holding it, and returns the address it wrote — empty when it wrote
// none, either because the report carried none or because the address was
// unusable.
//
// hostID is a predicate rather than a check made before the statement, the same
// idiom RecordVMNetworkIP and RecordVMHostKey use and for the same reason: a
// host may only ever speak for the VMs it holds, and a predicate leaves no
// read-then-write window in which the VM could move. A status reported for a VM
// that is not the reporter's own changes no rows and gets sql.ErrNoRows, the
// same answer a report for a row that is gone gets.
//
// An unusable address must NOT block the (status, last_error) transition: a
// genuinely-failed VM still needs status=failed persisted durably. So it is
// DROPPED — assigned_ip keeps its prior value via the CASE below — while the
// status still writes.
//
// The returned address is what makes that drop visible to the caller. A guard
// that silently blanks its argument and reports success is how syncsvc's status
// cache came to remember addresses that were never stored, and then suppress
// every later report that would have corrected them.
func (s *Store) RecordVMStatus(id, hostID, status, lastErr, ip string) (string, error) {
if ip != "" && !usableAddress(ip) {
ip = ""
}
res, err := s.db.Exec(
`UPDATE vms SET status=?, last_error=?, assigned_ip=CASE WHEN ?='' THEN assigned_ip ELSE ? END WHERE id=? AND host_id=?`,
status, lastErr, ip, ip, id, hostID,
)
if err != nil {
return "", err
}
n, _ := res.RowsAffected()
if n == 0 {
return "", sql.ErrNoRows
}
return ip, nil
}
// RecordVMNetworkIP persists the address a guest's named-network NIC was
// granted, as its host snooped it from the site's DHCP server.
//
// Separate from RecordVMStatus because it is a separate fact with a separate
// clock: assigned_ip exists the moment the reservation is made, while this one
// arrives whenever the site's server answers — which can be long before the VM
// is ready, the only phase RecordVMStatus writes in.
//
// The CASE is the assigned_ip rule, for the assigned_ip reason: an empty report
// is "not discovered", never "no longer has one", so it keeps what was stored
// rather than erasing it. An unusable address (link-local — what a guest shows
// when nothing answered — loopback, multicast, unparseable) is dropped the same
// way, so a NIC that never got a real lease reports nothing rather than noise.
//
// hostID is part of the WHERE clause rather than a check made before it, the
// same idiom RecordVMHostKey uses and for the same reason: a host may only
// ever speak for the VMs it holds, and a predicate leaves no read-then-write
// window in which the VM could move. A host that reports a lease for a VM
// that is not its own changes no rows and gets sql.ErrNoRows.
//
// network is also in the WHERE: a VM that never asked for a named network can
// never grow a LAN address, however an agent misbehaves — a forged report, or
// one from an agent whose view of the VM has skewed from the server's own.
// Refusing the write makes that pairing unrepresentable in the row rather
// than merely unrendered by the clients that know to hide it. This means
// sql.ErrNoRows now covers three predicates — the row is gone, the wrong host
// reported it, or the VM has no network — and the caller cannot tell which:
// a pre-check that classified the failure would reopen the read-then-write
// window this whole idiom exists to close. The resulting slog.Warn recurs at
// sync cadence by design (the write-through cache only commits on success, so
// it is never rate-limited away); it says only that something is wrong, and
// the VM id in the log line is what lets an operator resolve it in one query.
func (s *Store) RecordVMNetworkIP(id, hostID, ip string) error {
if ip != "" && !usableAddress(ip) {
ip = ""
}
res, err := s.db.Exec(
`UPDATE vms SET network_ip=CASE WHEN ?='' THEN network_ip ELSE ? END WHERE id=? AND host_id=? AND network != ''`,
ip, ip, id, hostID,
)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows
}
return nil
}
// RecordVMHostKey stores the public host key a host generated for one of its
// guests, together with the certificate the control plane signed for it.
//
// hostID is part of the WHERE clause rather than a check made before it. A host
// may only ever speak for the VMs it holds, and expressing that as a predicate
// instead of a read-then-write leaves no window between the two in which the
// VM could move. A host that reports a key for a VM that is not its own
// changes no rows and gets sql.ErrNoRows.
func (s *Store) RecordVMHostKey(vmID, hostID, pubkey, cert string) error {
res, err := s.db.Exec(
`UPDATE vms SET ssh_host_pubkey=?, ssh_host_cert=? WHERE id=? AND host_id=? AND deleted_at IS NULL`,
pubkey, cert, vmID, hostID,
)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows
}
return nil
}
// vmColumns is the positional column list every VM SELECT must use, so the
// order stays locked to scanVM's Scan below (which is positional, not
// name-based). queryVMs is the sole caller, so adding a column is a single
// edit here plus scanVM — every VM query goes through it and can't drift out
// of lockstep.
const vmColumns = `id, host_id, name, tenant, image_url, image_sha256, cloud_init, ssh_authorized_key,
ssh_host_pubkey, ssh_host_cert,
injected_key_type, injected_key_fp, injected_key_comment, trusted_cas,
vcpus, mem_mb, disk_gb, power_state, network, status, last_error, assigned_ip, network_ip,
created_at, deleted_at`
func scanVM(rows *sql.Rows) (VM, error) {
var vm VM
var createdAt string
var deletedAt sql.NullString
var trustedCAs sql.NullString
err := rows.Scan(
&vm.ID, &vm.HostID, &vm.Name, &vm.Tenant, &vm.ImageURL, &vm.ImageSHA256,
&vm.CloudInit, &vm.SSHAuthorizedKey, &vm.SSHHostPubKey, &vm.SSHHostCert,
&vm.InjectedKeyType, &vm.InjectedKeyFP, &vm.InjectedKeyComment, &trustedCAs,
&vm.VCPUs, &vm.MemMB, &vm.DiskGB,
&vm.PowerState, &vm.Network, &vm.Status, &vm.LastError, &vm.AssignedIP, &vm.NetworkIP,
&createdAt, &deletedAt,
)
if err != nil {
return VM{}, err
}
// NULL leaves TrustedCAs nil: the row was written before the column, and
// nothing here may invent a set it never recorded. Unparseable JSON is
// treated the same way for the same reason — an unreadable record is not a
// record, and guessing would be worse than admitting it.
if trustedCAs.Valid {
_ = json.Unmarshal([]byte(trustedCAs.String), &vm.TrustedCAs)
}
vm.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
if deletedAt.Valid {
t, _ := time.Parse(time.RFC3339, deletedAt.String)
vm.DeletedAt = &t
}
return vm, nil
}
// queryVMs runs a vmColumns-projected SELECT against vms, with where appended
// verbatim after `FROM vms` (e.g. " WHERE id=?", or "" for none) and args
// bound in order, scanning every matching row. listVMs, VMByTenantName, GetVM,
// and SpecForHost all share this — they differ only in WHERE clause,
// row-count expectations, and whether q is *sql.DB or an in-flight *sql.Tx.
func queryVMs(q querier, where string, args ...any) ([]VM, error) {
// Scan every row and CLOSE before reading volumes: the store runs one
// connection, so a second query issued while this result set is open would
// deadlock. scanVMRows is a separate function so the close is a scope
// boundary the compiler enforces rather than a rule to remember.
vms, err := scanVMRows(q, where, args...)
if err != nil {
return nil, err
}
// A VM's volumes are a list, so they live in their own table and are
// filled in here rather than being a column of the row above.
for i := range vms {
ids, err := volumeIDsForVM(q, vms[i].ID)
if err != nil {
return nil, fmt.Errorf("read vm volumes: %w", err)
}
vms[i].VolumeIDs = ids
}
return vms, nil
}
func scanVMRows(q querier, where string, args ...any) ([]VM, error) {
rows, err := q.Query(`SELECT `+vmColumns+` FROM vms`+where, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var vms []VM
for rows.Next() {
vm, err := scanVM(rows)
if err != nil {
return nil, err
}
vms = append(vms, vm)
}
return vms, rows.Err()
}
func (s *Store) ListVMs() ([]VM, error) { return listVMs(s.db) }
func listVMs(q querier) ([]VM, error) { return queryVMs(q, "") }
// VMByTenantName returns the live (non-tombstoned) VM with the given name
// WITHIN tenant. The vms_tenant_name unique index guarantees at most one
// match. Resolution is tenant-scoped by construction: a name in another
// tenant is sql.ErrNoRows, indistinguishable from absent. Used by the SSH
// jump gate to resolve `ssh -J gate user@<tenant>.<name>`.
func (s *Store) VMByTenantName(tenant, name string) (VM, error) {
vms, err := queryVMs(s.db, ` WHERE tenant=? AND name=? AND deleted_at IS NULL`, tenant, name)
if err != nil {
return VM{}, err
}
if len(vms) == 0 {
return VM{}, sql.ErrNoRows
}
return vms[0], nil
}
// GetVM returns the VM with the given id via a single indexed lookup on the
// primary key. Unlike VMByTenantName it does NOT filter on deleted_at: the
// patch/delete/restore callers operate on VMs that may be tombstoned, so the
// row must be found regardless of tombstone state. sql.ErrNoRows ⇒ no such VM.
func (s *Store) GetVM(id string) (VM, error) {
vms, err := queryVMs(s.db, ` WHERE id=?`, id)
if err != nil {
return VM{}, err
}
if len(vms) == 0 {
return VM{}, sql.ErrNoRows
}
return vms[0], nil
}
// Snapshot reads hosts, per-host allocation, and VMs in a single read
// transaction, so the trio is mutually consistent — a concurrent desired-state
// mutation between the reads cannot produce a payload mixing two epochs.
// The SSE stream builds its fleet snapshot from this.
func (s *Store) Snapshot() ([]Host, map[string]Alloc, []VM, error) {
tx, err := s.db.Begin()
if err != nil {
return nil, nil, nil, err
}
defer tx.Rollback()
hosts, err := listHosts(tx)
if err != nil {
return nil, nil, nil, err
}
alloc, err := allocatedByHost(tx)
if err != nil {
return nil, nil, nil, err
}
vms, err := listVMs(tx)
if err != nil {
return nil, nil, nil, err
}
return hosts, alloc, vms, tx.Commit()
}
// SpecForHost reads the desired VMs for one host plus the epoch they were read
// at, in a single transaction, so the pair cannot straddle a mutation. syncsvc
// renders the result into the pb.Snapshot that host is sent.
//
// Not a scoped Snapshot: that one reads the whole fleet (hosts, allocation,
// VMs) for the SSE stream. This is one host's spec, which is why it is named
// for what it returns rather than for the message built from it.
func (s *Store) SpecForHost(hostID string) (uint64, []VM, error) {
tx, err := s.db.Begin()
if err != nil {
return 0, nil, err
}
defer tx.Rollback()
var epoch uint64
if err := tx.QueryRow(`SELECT CAST(value AS INTEGER) FROM meta WHERE key='epoch'`).Scan(&epoch); err != nil {
return 0, nil, fmt.Errorf("read epoch: %w", err)
}
vms, err := queryVMs(tx, ` WHERE host_id=?`, hostID)
if err != nil {
return 0, nil, err
}
if err := tx.Commit(); err != nil {
return 0, nil, err
}
return epoch, vms, nil
}
// ServerCert returns the server's TLS cert PEM and its hex sha256 fingerprint,
// generating and persisting a self-signed cert on first call. Cert and key live
// beside the DB as server.crt / server.key.
func (s *Store) ServerCert() (certPEM []byte, fingerprint string, err error) {
certPath := filepath.Join(s.dbDir, "server.crt")
keyPath := filepath.Join(s.dbDir, "server.key")
certPEM, errC := os.ReadFile(certPath)
_, errK := os.ReadFile(keyPath)
if errC != nil || errK != nil {
var keyPEM []byte
certPEM, keyPEM, err = transport.GenerateServerCert()
if err != nil {
return nil, "", err
}
if err = os.WriteFile(certPath, certPEM, 0o600); err != nil {
return nil, "", err
}
if err = os.WriteFile(keyPath, keyPEM, 0o600); err != nil {
return nil, "", err
}
}
fp, err := transport.CertFingerprint(certPEM)
if err != nil {
return nil, "", err
}
return certPEM, fp, nil
}
// ServerKeyPEM returns the server key PEM (call after ServerCert has run).
func (s *Store) ServerKeyPEM() ([]byte, error) {
return os.ReadFile(filepath.Join(s.dbDir, "server.key"))
}