internal/server/syncsvc/syncsvc.go
Ref: Size: 33.6 KiB History
// Package syncsvc is the QUIC server end of the agent reconcile stream.
package syncsvc
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"sync"
"time"
"github.com/a73x/eitri/internal/pb"
"github.com/a73x/eitri/internal/server/hosttoken"
"github.com/a73x/eitri/internal/server/hub"
"github.com/a73x/eitri/internal/server/registry"
"github.com/a73x/eitri/internal/server/release"
"github.com/a73x/eitri/internal/server/store"
"github.com/a73x/eitri/internal/transport"
"github.com/quic-go/quic-go"
)
const defaultWriteTimeout = 30 * time.Second
// vmStatusRecorder is the durable-write seam applyReport uses to record VM
// lifecycle status. *store.Store satisfies it. Keeping it a field (rather than a
// direct s.st call) lets tests substitute a counting fake to prove that
// unchanged reports perform no write.
type vmStatusRecorder interface {
RecordVMStatus(id, hostID, status, lastErr, ip string) (string, error)
}
// Service is the QUIC server end of the agent reconcile stream.
type Service struct {
st *store.Store
reg *registry.Registry
hub *hub.Hub
secret []byte
// recorder is the durable VM-status write seam (defaults to st). tracker
// remembers the last durably-written status per VM so applyReport can skip
// RecordVMStatus (a SELECT+UPDATE on the single SQLite conn) when nothing
// changed.
recorder vmStatusRecorder
tracker *statusTracker
netTrack *netTracker
// certs signs a guest's reported host key.
certs hostCertSigner
// certTrack remembers the public key each VM was last certified for, so the
// steady state — every host repeating every VM's key every tick, forever —
// costs no database work. Same shape and same reason as netTrack.
certTrack *netTracker
// uplinkTrack remembers the uplink address each host last had WRITTEN, for
// the same reason netTrack does: every host reports every tick forever, and
// the store runs on a single connection.
uplinkTrack *netTracker
// netIPTrack remembers the named-network address each VM last had WRITTEN.
// Keyed by vmID like certTrack, and for the same reason: a bridged guest
// re-reports the address its lease renewed to, every tick, forever.
netIPTrack *netTracker
// maxCredAge, when non-zero, rejects credentials whose issued-at is older.
// Zero disables the age check (default: expiry without an auto-renewal
// channel would force periodic re-enrolls; per-host generation revocation
// is the primary mechanism, max-age is opt-in defense-in-depth).
maxCredAge time.Duration
// writeTimeout bounds each down-stream snapshot write (see defaultWriteTimeout).
writeTimeout time.Duration
// handshakeTimeout bounds the open/opened exchange on a server-initiated
// stream (see consoleHandshakeTimeout, its production value). A field, not
// the constant, so a test can inject a timeout it can actually outlive:
// nothing can prove the deadline is set — or cleared — against ten seconds.
handshakeTimeout time.Duration
// consoleMu guards conns: the live QUIC connection per agent, registered
// after auth in handleConn and deregistered when the session ends. The
// console broker opens per-session streams on it.
consoleMu sync.Mutex
conns map[string]quic.Connection
// offersMu guards offers: pending per-host agent self-upgrades, set by the
// API (operator click), carried in that host's snapshots, cleared when a
// Hello reports the target version. In-memory only — a restart forgets
// pending offers and the operator clicks again (idempotent).
offersMu sync.Mutex
offers map[string]offer
// now is the clock the offer timestamps are read from, injected the way
// registry.New takes one so a test can age an offer without sleeping.
now func() time.Time
}
// offer is one pending agent self-upgrade and the moment it was made. The
// timestamp is what lets a reader tell an upgrade in flight from one that has
// failed on the host: the work is a download, a checksum, a swap and a re-exec,
// so an offer still standing minutes later is not slow, it is stuck.
type offer struct {
up *pb.AgentUpgrade
offeredAt time.Time
}
// New constructs a Service with the production-default down-stream write
// timeout. maxCredAge zero disables the credential age check.
func New(st *store.Store, reg *registry.Registry, h *hub.Hub, secret []byte, maxCredAge time.Duration) *Service {
return newWithWriteTimeout(st, reg, h, secret, maxCredAge, defaultWriteTimeout)
}
// newWithWriteTimeout constructs a Service with an explicit down-stream write
// timeout. A zero timeout falls back to defaultWriteTimeout. Tests use this to
// inject a short timeout; New keeps the public signature unchanged.
func newWithWriteTimeout(st *store.Store, reg *registry.Registry, h *hub.Hub, secret []byte, maxCredAge, writeTimeout time.Duration) *Service {
if writeTimeout <= 0 {
writeTimeout = defaultWriteTimeout
}
return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout,
handshakeTimeout: consoleHandshakeTimeout,
conns: map[string]quic.Connection{}, recorder: st, tracker: newStatusTracker(), netTrack: newNetTracker(),
uplinkTrack: newNetTracker(), certTrack: newNetTracker(), netIPTrack: newNetTracker(),
offers: map[string]offer{}, now: time.Now}
}
// Serve accepts QUIC connections until ctx is cancelled.
func (s *Service) Serve(ctx context.Context, lis *quic.Listener) error {
for {
conn, err := lis.Accept(ctx)
if err != nil {
return err
}
go s.handleConn(ctx, conn)
}
}
// credStale evaluates a credential's claims against the host row's current
// generation: stale is true when the credential has been revoked (its
// generation no longer matches the host's current one) or has exceeded
// maxCredAge (expired, reported separately so callers that log an "expired"
// field can distinguish the reason). It is pure — no I/O — so callers decide
// when to fetch hostRow and what to do with a stale verdict (initial auth
// rejects; the per-report re-check closes the session).
func (s *Service) credStale(claims hosttoken.Claims, hostRow store.Host) (stale, expired bool) {
expired = s.maxCredAge > 0 && time.Since(claims.IssuedAt) > s.maxCredAge
stale = expired || claims.Generation != hostRow.CredGeneration
return stale, expired
}
// helloGrace bounds the unauthenticated head of a sync session, mirroring the
// SSH gate's handshakeGrace: a peer that completes the QUIC handshake but never
// sends Hello would otherwise park a goroutine for as long as it keeps the
// connection alive. A var (not const) so tests can shrink it.
var helloGrace = 30 * time.Second
// NOTE (verified Task 1): quic-go v0.48.2 uses INTERFACES quic.Connection and
// quic.Stream (not *quic.Conn/*quic.Stream, which only exist in v0.49+).
func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
// Every way out of here closes the connection. A handler that returns while
// leaving the transport up leaves an agent writing reports nobody reads —
// and QUIC keepalives hold that socket open indefinitely, so the host's last
// contact freezes and it reads offline until someone breaks the connection by
// hand. quic-go closes once (closeOnce), so the specific codes the auth paths
// below send still reach the agent; this only catches whatever they miss.
defer conn.CloseWithError(0, "session ended")
// Up-stream: agent opens it and sends Hello first. Both the accept and the
// first read run under the hello grace so an unauthenticated peer that
// handshakes then stalls cannot park this goroutine indefinitely.
acceptCtx, cancel := context.WithTimeout(ctx, helloGrace)
up, err := conn.AcceptStream(acceptCtx)
cancel()
if err != nil {
_ = conn.CloseWithError(transport.CodeAuthRejected, "no stream before hello grace")
return
}
_ = up.SetReadDeadline(time.Now().Add(helloGrace))
var first pb.AgentMessage
if err := transport.ReadMsg(up, &first, transport.DefaultMaxFrame); err != nil {
_ = conn.CloseWithError(transport.CodeAuthRejected, "no hello before grace")
return
}
// Hello received — clear the deadline so it never applies to the long-lived
// report stream that follows.
_ = up.SetReadDeadline(time.Time{})
h := first.GetHello()
if h == nil {
_ = conn.CloseWithError(transport.CodeAuthRejected, "first frame must be Hello")
return
}
cred := h.GetCredential()
// host_id in Hello is advisory; the authenticated identity comes from the credential (hosttoken.Verify), so a spoofed Hello.HostId cannot mislead us.
claims, ok := hosttoken.Verify(s.secret, cred)
if !ok {
_ = conn.CloseWithError(transport.CodeAuthRejected, "invalid host credential")
return
}
hostID := claims.HostID
hostRow, err := s.st.GetHost(hostID)
if err != nil {
_ = conn.CloseWithError(transport.CodeAuthRejected, "host not found")
return
}
// credStale covers both per-host revocation (a credential minted at an
// older generation is dead) and max-age expiry in a single predicate; the
// close message distinguishes the two so an operator reading the close
// reason can tell a routine re-enroll (expiry) from something more
// suspicious (revocation). The ordering is deliberate: host-not-found is
// checked BEFORE expiry, so it wins for a purged host — earlier code
// rejected an expired credential before ever touching the DB.
if stale, expired := s.credStale(claims, hostRow); stale {
msg := "credential revoked — re-enroll this host"
if expired {
msg = "credential expired — re-enroll this host"
}
_ = conn.CloseWithError(transport.CodeAuthRejected, msg)
return
}
s.reg.RecordConnect(hostID)
s.reg.SetAgentVersion(hostID, h.GetFacts().GetAgentVersion())
s.reg.SetHostNetworks(hostID, h.GetHostNetworks())
s.clearOfferIfDone(hostID, h.GetFacts().GetAgentVersion())
slog.Info("agent connected", "host", hostID, "provisioner", h.GetProvisioner(), "last_seen_epoch", h.GetLastSeenEpoch())
// Best-effort: refresh the host's OS facts from this Hello. A failed write
// must not drop the connection — the report path is otherwise authoritative.
if err := s.st.UpdateHostFacts(hostID, toStoreFacts(h)); err != nil {
slog.Warn("update host facts", "host", hostID, "err", err)
}
// Down-stream: server opens it; first write makes it visible to the agent.
down, err := conn.OpenStreamSync(ctx)
if err != nil {
return
}
// Console reachability: only now — with the down-stream open — is it safe
// for OpenConsole to add streams to this connection (stream-order
// invariant: the snapshot down-stream is always the first accepted).
s.consoleMu.Lock()
displaced := s.conns[hostID]
s.conns[hostID] = conn
s.consoleMu.Unlock()
if displaced != nil {
// Outside the lock: CloseWithError waits for the displaced connection to
// finish tearing down, and the console broker must not queue behind that.
slog.Info("newer session for this host; closing the older one", "host", hostID)
_ = displaced.CloseWithError(transport.CodeSuperseded, "superseded by a newer session for this host")
}
defer func() {
s.consoleMu.Lock()
if s.conns[hostID] == conn {
delete(s.conns, hostID)
}
s.consoleMu.Unlock()
// Forget this host's cached network facts, so the next connection writes
// them once rather than trusting a memory of a row that may have been
// removed, re-enrolled or edited while the host was away. Bounds the maps
// to connected hosts, and costs one write per reconnect.
s.netTrack.forget(hostID)
s.uplinkTrack.forget(hostID)
}()
// Single writer for the down-stream: the poke goroutine.
pokes, cancel := s.hub.Subscribe(hostID)
defer cancel()
sendErr := make(chan error, 1)
go func() {
if err := s.pushSnapshot(down, hostID); err != nil {
s.failWrite(conn, hostID, err)
sendErr <- err
return
}
for range pokes {
if err := s.pushSnapshot(down, hostID); err != nil {
s.failWrite(conn, hostID, err)
sendErr <- err
return
}
}
sendErr <- nil
}()
// Read loop: up-stream reports only.
for {
var msg pb.AgentMessage
if err := transport.ReadMsg(up, &msg, transport.DefaultMaxFrame); err != nil {
if !errors.Is(err, io.EOF) {
slog.Warn("agent stream ended", "host", hostID, "err", err)
}
return
}
if rep := msg.GetReport(); rep != nil {
row, err := s.st.GetHost(hostID)
if err != nil {
// Transient store failure — NOT an auth verdict. Close with a
// plain code so the agent retries on its normal 5s backoff
// instead of the permanent-auth 60s path.
slog.Warn("credential re-check failed; dropping session", "host", hostID, "err", err)
_ = conn.CloseWithError(0, "credential re-check unavailable")
return
}
if stale, expired := s.credStale(claims, row); stale {
slog.Info("credential no longer valid mid-session; closing", "host", hostID, "expired", expired)
_ = conn.CloseWithError(transport.CodeAuthRejected, "credential revoked or expired — re-enroll this host")
return
}
s.applyReport(hostID, rep)
}
select {
case e := <-sendErr:
if e != nil {
slog.Warn("down-stream push failed", "host", hostID, "err", e)
}
return
default:
}
}
}
// buildSnapshot reads hostID's desired state — its VMs and the exposures it
// should be serving — and renders it as one snapshot.
func (s *Service) buildSnapshot(hostID string) (*pb.Snapshot, error) {
epoch, vms, err := s.st.SpecForHost(hostID)
if err != nil {
return nil, fmt.Errorf("desired for host: %w", err)
}
snap := &pb.Snapshot{Epoch: epoch, Vms: make([]*pb.VMSpec, 0, len(vms))}
snap.AgentUpgrade = s.offerFor(hostID)
caCache := map[string][]string{} // tenant -> canonical CA lines, legacy rows only
for _, v := range vms {
cas := make([]string, 0, len(v.TrustedCAs))
for _, c := range v.TrustedCAs {
cas = append(cas, c.AuthorizedKey)
}
// This is now vestigial: store.Open freezes a set onto every unrecorded
// row it finds, so by the time a snapshot is built there are none left.
// It survives as a safety net for the one row the backfill cannot have
// seen — a VM created by a pre-upgrade server still running alongside
// this one, mid-rollout. For that one the fallback preserves the old
// behaviour and the guest works, where an empty set would seed a guest
// that trusts no CA at all and is unreachable for good. It must NOT
// become the way legacy VMs are served again: a live set here is a
// second authority for a fact the row is supposed to own, and a guest
// fed by it can never be shown as trusting a stale CA.
if v.TrustedCAs == nil {
legacy, ok := caCache[v.Tenant]
if !ok {
list, err := s.st.ListTenantUserCAs(v.Tenant)
if err != nil {
return nil, fmt.Errorf("list tenant user cas: %w", err)
}
legacy = make([]string, 0, len(list))
for _, c := range list {
legacy = append(legacy, c.Pubkey)
}
caCache[v.Tenant] = legacy
}
cas = legacy
}
snap.Vms = append(snap.Vms, &pb.VMSpec{
VmId: v.ID, Name: v.Name, ImageUrl: v.ImageURL, ImageSha256: v.ImageSHA256,
CloudInit: v.CloudInit, Vcpus: v.VCPUs, MemMb: v.MemMB, DiskGb: v.DiskGB,
Persistent: true, PowerState: v.PowerState, Tombstoned: v.DeletedAt != nil,
SshAuthorizedKey: v.SSHAuthorizedKey,
SshUserCaAuthorizedKeys: cas,
SshHostCert: v.SSHHostCert,
// Which underlay this guest attaches to: a named host network, or
// "" for the NAT default. Admission has already proven the host
// advertises the name, so the agent is only being told what it
// said it could serve.
Network: v.Network,
// The volumes this guest attaches, in the order it named their
// claims. The agent materialises each file before boot; an agent
// too old to read this field is kept away from such a VM by the
// snapshot floor below.
VolumeIds: v.VolumeIDs,
// With a CA in hand the fleet issues host certificates, so a guest
// must present one. The host generates the key, reports the public
// half, and holds the guest at the gate until the certificate for
// it comes back down in a later snapshot.
HostCertRequired: s.certs != nil,
})
}
exps, err := s.st.ListExposuresForHost(hostID)
if err != nil {
return nil, fmt.Errorf("list host exposures: %w", err)
}
for _, e := range exps {
// In-range by construction: the API validates ports and closes the
// protocol at the two the agent can bind (validateExposure), and the
// store accepts rows only from the API.
snap.Exposures = append(snap.Exposures, &pb.ExposureSpec{
Id: e.ID, VmId: e.VMID,
GuestPort: uint32(e.GuestPort), HostPort: uint32(e.HostPort),
Protocol: e.Protocol,
})
}
// Tombstoned rows ride along with the live ones: reclaiming a file is work
// the agent can only do while it is still told the file exists.
vols, err := s.st.ListVolumesForHost(hostID)
if err != nil {
return nil, fmt.Errorf("list host volumes: %w", err)
}
for _, v := range vols {
snap.Volumes = append(snap.Volumes, &pb.VolumeSpec{
VolumeId: v.ID, SizeGb: v.SizeGB, Tombstoned: v.DeletedAt != nil,
})
}
// The floor is raised only by a snapshot that uses the feature: a host
// with no volumes keeps serving a pre-volumes agent.
if len(snap.Volumes) > 0 {
snap.MinAgentVersion = release.Volumes.Since
}
return snap, nil
}
// pushSnapshot sends hostID's current desired state down the stream.
func (s *Service) pushSnapshot(down quic.Stream, hostID string) error {
snap, err := s.buildSnapshot(hostID)
if err != nil {
return err
}
if err := down.SetWriteDeadline(time.Now().Add(s.writeTimeout)); err != nil {
return fmt.Errorf("set write deadline: %w", err)
}
return transport.WriteMsg(down, &pb.ServerMessage{Msg: &pb.ServerMessage_Snapshot{Snapshot: snap}})
}
// OfferAgentUpgrade records a pending agent self-upgrade for hostID; the
// host's next snapshot carries it (callers poke the host via the hub). A second
// click re-offers and restarts the clock: the operator has asked again, and the
// wait they are being told about is the wait since that ask.
func (s *Service) OfferAgentUpgrade(hostID, version, url, sha256 string) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
s.offers[hostID] = offer{
up: &pb.AgentUpgrade{Version: version, Url: url, Sha256: sha256},
offeredAt: s.now(),
}
}
// offerFor returns hostID's pending upgrade (nil when none).
func (s *Service) offerFor(hostID string) *pb.AgentUpgrade {
s.offersMu.Lock()
defer s.offersMu.Unlock()
return s.offers[hostID].up
}
// PendingAgentUpgrade reports hostID's outstanding offer: the version offered
// and how long it has been standing. ok is false when there is none.
//
// Offers live in memory only. A server restart forgets them, so the pending
// state disappears and the button comes back — which is the honest answer,
// because a restarted server has also forgotten to put the offer in the next
// snapshot. The cure is the same click as before.
func (s *Service) PendingAgentUpgrade(hostID string) (version string, age time.Duration, ok bool) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
o, ok := s.offers[hostID]
if !ok {
return "", 0, false
}
return o.up.GetVersion(), s.now().Sub(o.offeredAt), true
}
func (s *Service) ClearAgentUpgrade(hostID string) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
delete(s.offers, hostID)
}
func (s *Service) clearOfferIfDone(hostID, reportedVersion string) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
if o, ok := s.offers[hostID]; ok && o.up.GetVersion() == reportedVersion {
delete(s.offers, hostID)
}
}
// failWrite handles a failed down-stream write by closing the connection. A
// canceled hub subscription does NOT unblock an in-flight Write, and the read
// loop is parked in ReadMsg(up); closing the connection unblocks that ReadMsg so
// handleConn returns and runs its cleanup (cancel the hub subscription).
func (s *Service) failWrite(conn quic.Connection, hostID string, err error) {
slog.Warn("down-stream write failed; closing connection", "host", hostID, "err", err)
// Close code 0 (not CodeAuthRejected): a write timeout is a transport/liveness
// problem, not a permanent auth failure, so the agent should reconnect with
// normal backoff rather than treat its credential as dead.
_ = conn.CloseWithError(0, "down-stream write timeout")
}
// applyReport updates the registry and durably records VM status changes.
// Errors within the report are logged and skipped — they must never kill the stream.
func (s *Service) applyReport(hostID string, rep *pb.Report) {
// Build registry report.
r := registry.Report{
LastSeenEpoch: rep.GetLastSeenEpoch(),
FenceViolation: rep.GetFenceViolation(),
}
r.VMs = toRegistryVMs(rep.GetVms())
r.Quarantined = toRegistryQuarantined(rep.GetQuarantined())
r.Exposures = toRegistryExposures(rep.GetExposures())
r.Volumes = toRegistryVolumes(rep.GetVolumes())
r.Capacity = toRegistryCapacity(rep.GetCapacity())
r.Metrics = toRegistryMetrics(rep.GetMetrics())
s.reg.UpdateReport(hostID, r)
s.certifyReportedHostKeys(hostID, rep.GetVms())
for _, v := range rep.GetVms() {
vmID := v.GetVmId()
if netIP := v.GetNetworkIp(); netIP != "" {
if err := s.netIPTrack.writeThrough(vmID, netIP, func() error {
return s.st.RecordVMNetworkIP(vmID, hostID, netIP)
}); err != nil {
slog.Warn("record vm network ip", "vm", vmID, "host", hostID, "ip", netIP, "err", err)
}
}
phase := v.GetPhase()
if phase != "ready" && phase != "failed" {
continue
}
err := s.tracker.writeThrough(vmID, phase, v.GetLastError(), v.GetIp(), func() (string, error) {
return s.recorder.RecordVMStatus(vmID, hostID, phase, v.GetLastError(), v.GetIp())
})
if err != nil {
slog.Warn("RecordVMStatus rejected", "vm", vmID, "host", hostID, "err", err)
}
}
if cidr := rep.GetGuestCidr(); cidr != "" {
if err := s.netTrack.writeThrough(hostID, cidr, func() error {
return s.st.RecordHostNetwork(hostID, cidr)
}); err != nil {
slog.Warn("record host network", "host", hostID, "cidr", cidr, "err", err)
}
}
if addr := rep.GetHostUplinkAddr(); addr != "" {
if err := s.uplinkTrack.writeThrough(hostID, addr, func() error {
return s.st.RecordHostUplink(hostID, addr)
}); err != nil {
slog.Warn("record host uplink", "host", hostID, "addr", addr, "err", err)
}
}
// Fence violation: log ERROR and point at the restore runbook.
if rep.GetFenceViolation() {
slog.Error("agent refused snapshot: epoch fence violation — see restore runbook",
"host", hostID, "agent_epoch", rep.GetLastSeenEpoch())
}
// A reaped VM shares its host's tenant; resolve it once so the terminal
// vm.reap row is tenant-scoped like the rest of the VM's timeline. The host
// row still exists during graceful reap (RemoveHost waits for the drain);
// fall back to the system audit scope only for the defensive race where it
// is gone.
destroyed := rep.GetDestroyed()
reapTenant := store.SystemTenant
if len(destroyed) > 0 {
if h, err := s.st.GetHost(hostID); err == nil {
reapTenant = h.Tenant
}
}
anyDeleted := false
for _, id := range destroyed {
// Prune the dedup cache: this VM's row is being hard-deleted, so its
// tracked status is dead weight (and a future id reuse must not inherit
// a stale cached triple — ids are unique, but forgetting is the correct,
// memory-bounding thing regardless of whether the delete succeeds).
s.tracker.forget(id)
s.certTrack.forget(id)
s.netIPTrack.forget(id)
if err := s.st.HardDeleteVM(id, hostID); err != nil {
slog.Warn("HardDeleteVM failed", "vm", id, "host", hostID, "err", err)
} else {
anyDeleted = true
// Terminal lifecycle event. The VM row is already gone, so only the
// id and host survive — vm_id is the key the per-VM timeline filters
// on. Best-effort: a failed audit must not break the reap.
detail, _ := json.Marshal(map[string]string{
"vm_id": id, "host_id": hostID, "reason": "destroyed after tombstone grace",
})
if err := s.st.AppendAudit(reapTenant, "vm.reap", string(detail)); err != nil {
slog.Warn("audit vm.reap failed", "vm", id, "host", hostID, "err", err)
}
}
}
if anyDeleted {
s.hub.Poke(hostID)
}
// A fenced report is the agent saying it refused this snapshot, so nothing
// in it is an answer to what the snapshot asked for. It must not drive a
// delete: the reap reads an omitted volume as gone, and a report from a
// host that acted on nothing would reap every tombstoned row on it.
if !rep.GetFenceViolation() {
s.reapVolumes(hostID, rep.GetVolumes())
}
}
// reapVolumes hard-deletes a tombstoned volume once its host reports the file
// gone: present=false, or absent from a report by an agent that reports
// volumes at all. A live volume is never touched here whatever the report
// says — the row is the truth the agent converges toward, and a file it has
// not made yet is a converge still owed, not a row to delete — and a
// pre-volumes agent's silence proves nothing, so it reaps nothing. A fenced
// report reaches none of this: see the caller.
func (s *Service) reapVolumes(hostID string, reported []*pb.VolumeStatus) {
hs, ok := s.reg.Get(hostID)
if !ok || !release.Volumes.SupportedBy(hs.AgentVersion) {
return
}
vols, err := s.st.ListVolumesForHost(hostID)
if err != nil {
slog.Warn("list host volumes", "host", hostID, "err", err)
return
}
present := map[string]bool{}
for _, v := range reported {
present[v.GetVolumeId()] = v.GetPresent()
}
reaped := false
for _, v := range vols {
if v.DeletedAt == nil || present[v.ID] {
continue
}
// The claim is tombstoned by now, so read it whatever its state: the
// terminal audit row belongs in the tenant's timeline, and the system
// scope is only for the race where the row has already gone.
tenant := store.SystemTenant
if c, err := s.st.GetVolumeClaimAny(v.ClaimID); err == nil {
tenant = c.Tenant
}
// Refused unless the claim is tombstoned too — the store's guard, not
// this loop's assumption. See HardDeleteVolume.
if err := s.st.HardDeleteVolume(v.ID); err != nil {
slog.Warn("reap volume", "volume", v.ID, "host", hostID, "err", err)
continue
}
reaped = true
detail, _ := json.Marshal(map[string]string{"volume_id": v.ID, "host_id": hostID, "claim_id": v.ClaimID})
if err := s.st.AppendAudit(tenant, "volume.reap", string(detail)); err != nil {
slog.Warn("audit volume.reap failed", "volume", v.ID, "host", hostID, "err", err)
}
}
// The agent still holds a snapshot listing the tombstone and would re-report
// it every tick; poke it once so the next snapshot is free of it.
if reaped {
s.hub.Poke(hostID)
}
}
// toRegistryVolumes maps reported volume state to registry rows. Returns nil
// (not an empty slice) for empty input, matching append-into-nil behavior.
func toRegistryVolumes(in []*pb.VolumeStatus) []registry.VolumeStatus {
if len(in) == 0 {
return nil
}
out := make([]registry.VolumeStatus, 0, len(in))
for _, v := range in {
out = append(out, registry.VolumeStatus{VolumeID: v.GetVolumeId(), Present: v.GetPresent(), SizeGB: v.GetSizeGb()})
}
return out
}
// toRegistryVMs maps reported VMStatus rows to registry rows. Returns nil (not
// an empty slice) for empty input, matching the original append-into-nil
// behavior.
func toRegistryVMs(in []*pb.VMStatus) []registry.VMStatus {
if len(in) == 0 {
return nil
}
out := make([]registry.VMStatus, 0, len(in))
for _, v := range in {
out = append(out, registry.VMStatus{
VMID: v.GetVmId(),
PowerState: v.GetPowerState(),
Phase: v.GetPhase(),
IP: v.GetIp(),
LastError: v.GetLastError(),
StatusDetail: v.GetStatusDetail(),
})
}
return out
}
// toRegistryQuarantined maps reported quarantined VMs to registry rows. Returns
// nil (not an empty slice) for empty input, matching append-into-nil behavior.
func toRegistryQuarantined(in []*pb.QuarantinedVM) []registry.QuarantinedVM {
if len(in) == 0 {
return nil
}
out := make([]registry.QuarantinedVM, 0, len(in))
for _, q := range in {
out = append(out, registry.QuarantinedVM{
VMID: q.GetVmId(),
Name: q.GetName(),
VMSpecJSON: q.GetVmspecJson(),
DestroyAtUnix: q.GetDestroyAtUnix(),
})
}
return out
}
// toRegistryExposures maps reported exposure state to registry rows. Returns
// nil (not an empty slice) for empty input, matching append-into-nil behavior.
func toRegistryExposures(in []*pb.ExposureStatus) []registry.ExposureStatus {
if len(in) == 0 {
return nil
}
out := make([]registry.ExposureStatus, 0, len(in))
for _, e := range in {
out = append(out, registry.ExposureStatus{
ID: e.GetId(), State: e.GetState(), Reason: e.GetReason(),
Sessions: toRegistrySessions(e.GetSessions()),
})
}
return out
}
func toRegistrySessions(s *pb.ExposureSessions) *registry.ExposureSessions {
if s == nil {
return nil
}
return ®istry.ExposureSessions{
Active: s.GetActive(), Refused: s.GetRefused(), Dropped: s.GetDropped(),
}
}
// toRegistryCapacity maps reported capacity (nil → zero value).
func toRegistryCapacity(c *pb.Capacity) registry.Capacity {
if c == nil {
return registry.Capacity{}
}
return registry.Capacity{VCPUs: c.GetVcpus(), MemMB: c.GetMemMb(), DiskGB: c.GetDiskGb()}
}
// toStoreFacts maps what a Hello says about its host to the store shape. The
// provisioner is a top-level Hello field rather than one of HostFacts, but it
// is the same KIND of fact — slow-changing host identity, refreshed on every
// reconnect — so it is written by the same path instead of a second one.
func toStoreFacts(h *pb.Hello) store.HostFacts {
f := h.GetFacts()
return store.HostFacts{
OSID: f.GetOsId(), OSPretty: f.GetOsPretty(), OSVersion: f.GetOsVersion(),
Kernel: f.GetKernel(), CPUModel: f.GetCpuModel(), Virt: f.GetVirt(),
Provisioner: h.GetProvisioner(),
}
}
// toRegistryMetrics maps reported live metrics (nil → zero value).
func toRegistryMetrics(m *pb.HostMetrics) registry.Metrics {
if m == nil {
return registry.Metrics{}
}
return registry.Metrics{
UptimeS: m.GetUptimeS(),
MemUsedMB: m.GetMemUsedMb(),
MemAvailableMB: m.GetMemAvailableMb(),
Load1: m.GetLoad1(),
Load5: m.GetLoad5(),
Load15: m.GetLoad15(),
DiskUsedGB: m.GetDiskUsedGb(),
DiskFreeGB: m.GetDiskFreeGb(),
}
}
// ErrAgentOffline reports that the target host has no live sync connection.
var ErrAgentOffline = errors.New("agent not connected")
// consoleHandshakeTimeout is the production value of Service.handshakeTimeout:
// it bounds the ConsoleOpen/ConsoleOpened exchange so a wedged agent cannot pin
// the WS handler. The bridged session itself has no deadline — consoles are
// long-lived.
const consoleHandshakeTimeout = 10 * time.Second
// openStream is the shared server-initiated stream procedure behind OpenConsole
// and OpenTCP: find the live agent conn, open a QUIC stream, send `open` under a
// handshake deadline, await the reply, and return the stream as a raw byte pipe
// once the agent acks. checkReply extracts the typed ack — present=false when
// the reply wasn't the expected ack message, ok=false with a reason when the
// agent refused. label ("console"/"tcp") prefixes the error messages. The
// returned Close tears down both directions.
//
// ctx bounds stream OPENING only; the handshake that follows is bounded by
// s.handshakeTimeout instead, so a call can outlive ctx cancellation by up to
// that long (10s in production) before returning.
func (s *Service) openStream(ctx context.Context, hostID, label string, open *pb.ServerMessage,
checkReply func(*pb.AgentMessage) (ok bool, reason string, present bool)) (io.ReadWriteCloser, error) {
s.consoleMu.Lock()
conn, ok := s.conns[hostID]
s.consoleMu.Unlock()
if !ok {
return nil, ErrAgentOffline
}
st, err := conn.OpenStreamSync(ctx)
if err != nil {
return nil, fmt.Errorf("open %s stream: %w", label, err)
}
cs := consoleStream{st}
if err := st.SetDeadline(time.Now().Add(s.handshakeTimeout)); err != nil {
cs.Close()
return nil, err
}
if err := transport.WriteMsg(st, open); err != nil {
cs.Close()
return nil, fmt.Errorf("%s open: %w", label, err)
}
var reply pb.AgentMessage
if err := transport.ReadMsg(st, &reply, transport.DefaultMaxFrame); err != nil {
cs.Close()
return nil, fmt.Errorf("%s reply: %w", label, err)
}
replyOK, reason, present := checkReply(&reply)
if !present {
cs.Close()
return nil, fmt.Errorf("%s refused: unexpected reply", label)
}
if !replyOK {
cs.Close()
return nil, fmt.Errorf("%s refused: %s", label, reason)
}
// Handshake done — clear the deadline; the session is long-lived.
if err := st.SetDeadline(time.Time{}); err != nil {
cs.Close()
return nil, err
}
return cs, nil
}
// OpenConsole opens a console stream to vmID's agent on the live sync
// connection: sends ConsoleOpen, awaits ConsoleOpened, and returns the stream
// as a raw byte pipe.
func (s *Service) OpenConsole(ctx context.Context, hostID, vmID string) (io.ReadWriteCloser, error) {
open := &pb.ServerMessage{Msg: &pb.ServerMessage_ConsoleOpen{ConsoleOpen: &pb.ConsoleOpen{VmId: vmID}}}
return s.openStream(ctx, hostID, "console", open, func(m *pb.AgentMessage) (bool, string, bool) {
co := m.GetConsoleOpened()
if co == nil {
return false, "", false
}
return co.GetOk(), co.GetError(), true
})
}
// OpenTCP opens a tunnel stream to vmID's agent on the live sync connection:
// sends TCPOpen (naming the VM and guest TCP port), awaits TCPOpened, and
// returns the stream as a raw byte pipe.
func (s *Service) OpenTCP(ctx context.Context, hostID, vmID string, port uint32) (io.ReadWriteCloser, error) {
open := &pb.ServerMessage{Msg: &pb.ServerMessage_TcpOpen{TcpOpen: &pb.TCPOpen{VmId: vmID, Port: port}}}
return s.openStream(ctx, hostID, "tcp", open, func(m *pb.AgentMessage) (bool, string, bool) {
to := m.GetTcpOpened()
if to == nil {
return false, "", false
}
return to.GetOk(), to.GetError(), true
})
}
// consoleStream adapts a quic.Stream to io.ReadWriteCloser with a Close that
// tears down BOTH directions (Stream.Close only closes the write side).
type consoleStream struct{ quic.Stream }
func (c consoleStream) Close() error {
c.CancelRead(0)
return c.Stream.Close()
}