internal/agent/syncclient/client.go
Ref: Size: 25.5 KiB History
// Package syncclient holds the agent's stream loop: receive snapshots,
// run engine steps, send reports. Reconnects with backoff forever.
package syncclient
import (
"context"
"errors"
"io"
"log/slog"
"net"
"os"
"runtime"
"strings"
"sync"
"sync/atomic"
"time"
agentexec "github.com/a73x/eitri/internal/agent/exec"
"github.com/a73x/eitri/internal/agent/hostinfo"
"github.com/a73x/eitri/internal/agent/reconcile"
"github.com/a73x/eitri/internal/agent/selfupdate"
"github.com/a73x/eitri/internal/agent/state"
"github.com/a73x/eitri/internal/pb"
"github.com/a73x/eitri/internal/transport"
"github.com/a73x/eitri/internal/version"
"github.com/quic-go/quic-go"
)
// computeCapacity is the raw-capacity source, indirected through a var so tests
// can count how often the (memoized) computation actually runs.
var computeCapacity = hostinfo.Capacity
// Console bridges server-opened console streams to a VM's serial pump
// (consumer-owned; the concrete implementation is *serialpump.Manager).
// onReady fires after validation and before any bytes — the accept loop uses
// it to send the ConsoleOpened ok-frame, so a refusal (unknown VM) can still
// be reported as ok=false. onReady MUST be invoked synchronously — before
// Attach returns, on the caller's goroutine — because the accept loop's
// sentReady bookkeeping depends on that ordering. nil Console refuses every
// console request.
type Console interface {
Attach(ctx context.Context, vmID string, rw io.ReadWriter, onReady func() error) error
}
// Exposures converges this host's published guest ports and reports what each
// listener is doing (consumer-owned; the concrete implementation is
// *exposeproxy.Manager). A nil Exposures converges nothing and reports nothing
// — an agent with no proxy publishes no ports, and says so by saying nothing.
type Exposures interface {
Converge(desired []*pb.ExposureSpec) []*pb.ExposureStatus
}
// DefaultTickInterval is the fallback report/reconcile cadence when
// Client.TickInterval is zero. It is COUPLED to the server's
// registry.OnlineWindow (30s): the server marks a host offline after
// OnlineWindow of silence, so the agent must report well inside it. The invariant
// OnlineWindow >= 3*DefaultTickInterval gives a ~3-missed-report margin before a
// healthy host flaps Online/Offline; changing either value alone erodes it (see
// registry.OnlineWindow and the invariant test in this package). Independent of
// transport.SyncKeepAlivePeriod, which keeps the QUIC connection itself alive.
const DefaultTickInterval = 10 * time.Second
// Client manages the agent's QUIC sync session.
type Client struct {
Engine *reconcile.Engine
St *state.Store
Identity state.Identity
StateDir string
// Provisioner is the VM backend this host runs, echoed in Hello so the
// server's connection log says what kind of host arrived. The value the
// server stores comes from enroll, not from here.
Provisioner string
// GuestCIDR reports the subnet this host's guests are on, asked once per
// report. Nil, or an empty return, means "no answer" — never "no network".
GuestCIDR func() string
// HostNetworks are the named guest networks this host is configured to
// serve (--host-network), advertised in Hello rather than in the report:
// the set changes only with the agent's command line, i.e. a restart. A
// host that advertises none takes no VM that asks for a network by name.
HostNetworks []string
// Runner executes host-introspection subprocesses (on Linux,
// systemd-detect-virt via hostinfo). Injected so this data-plane package
// never imports os/exec (R6); nil is tolerated (virt reported as unknown).
Runner agentexec.Runner
// Console handles server-opened console streams (nil refuses them all).
Console Console
// Exposures publishes this host's guest ports (nil publishes none).
Exposures Exposures
// UplinkAddr reports the address this host answers on, asked once per
// report like GuestCIDR. Nil, or an empty return, means "no answer".
UplinkAddr func() string
// dialGuest connects to a VM's ssh port; overridable in tests. nil uses the
// production dialer, which pins the guest port to 22 — the wire port is
// validated but never dialed, so a compromised control plane cannot redirect
// the tunnel to an arbitrary port.
dialGuest func(ip string) (net.Conn, error)
// applyUpgrade performs the self-update (overridable in tests). nil uses
// a default selfupdate.Applier; on success it never returns (re-exec).
applyUpgrade func(ctx context.Context, u selfupdate.Update) error
// upgrading guards one self-update attempt in flight across snapshots.
upgrading atomic.Bool
// TickInterval is the period of the fallback ticker that drives a reconcile
// step even when no new snapshot has arrived (e.g. for periodic health
// reports). Zero uses DefaultTickInterval (coupled to registry.OnlineWindow).
TickInterval time.Duration
// ReconnectBackoff is the sleep between a transient session failure and the
// next dial attempt. Zero uses the production default of 5 seconds.
ReconnectBackoff time.Duration
// MaxVCPUs, MaxMemMB, MaxDiskGB cap the capacity this agent advertises to
// the fleet (0 = unlimited). They let an operator reserve host headroom
// rather than donating the whole machine. Enforcement of the same caps at
// VM-boot time lives on reconcile.Engine; this is the advertised half.
MaxVCPUs int64
MaxMemMB int64
MaxDiskGB int64
// rawCap memoizes the machine's real capacity (vCPUs/RAM/disk). Host totals
// don't change over a session, so the host is probed once instead of on
// every report; the cheap per-report clamp still applies. Only a FULL
// reading is cached — capacity() silently zeroes a dimension whose probe
// failed, and caching that would freeze a bad advertisement for the whole
// session, so a zeroed probe is re-tried on the next call. Guarded by
// rawCapMu (hello and report goroutines both call advertisedCapacity).
rawCapMu sync.Mutex
rawCap *pb.Capacity
}
// clampCapacity reduces each advertised dimension to its configured cap
// (0 = unlimited). It never inflates: a cap above the real total is a no-op, so
// an agent cannot advertise more than the machine actually has.
func clampCapacity(cap *pb.Capacity, maxVCPUs, maxMemMB, maxDiskGB int64) *pb.Capacity {
return &pb.Capacity{
Vcpus: clampDim(cap.GetVcpus(), maxVCPUs),
MemMb: clampDim(cap.GetMemMb(), maxMemMB),
DiskGb: clampDim(cap.GetDiskGb(), maxDiskGB),
}
}
func clampDim(actual, cap int64) int64 {
if cap > 0 && cap < actual {
return cap
}
return actual
}
// advertisedCapacity is the machine's real capacity clamped to this agent's
// configured caps — what the agent reports to the server. The raw capacity is
// computed once (host totals are fixed for the session) and re-clamped cheaply
// on every call.
func (c *Client) advertisedCapacity(stateDir string) *pb.Capacity {
c.rawCapMu.Lock()
raw := c.rawCap
if raw == nil {
raw = computeCapacity(stateDir)
// Cache only a full reading. A zeroed dimension means the Statfs/Sysinfo
// probe failed; caching it would freeze a bad total for the session, so
// leave rawCap nil and re-probe next call (vCPUs is runtime.NumCPU, never
// zero, so only mem/disk gate the memoization).
if raw.GetMemMb() > 0 && raw.GetDiskGb() > 0 {
c.rawCap = raw
}
}
c.rawCapMu.Unlock()
return clampCapacity(raw, c.MaxVCPUs, c.MaxMemMB, c.MaxDiskGB)
}
// maybeUpgrade launches the agent self-update named by a snapshot's
// AgentUpgrade. No-op when the target equals the running version (the offer
// has converged) or an attempt is already in flight. On success the process
// re-execs and never returns; on failure the flight is released and the next
// snapshot carrying the offer retries — level-triggered like everything else.
//
// ctx here is deliberately session()'s OUTER parameter, not the per-session
// sessCtx: the download must survive a mere reconnect (a QUIC blip must not
// abandon a half-downloaded artifact only to restart it from scratch next
// session), and c.upgrading is a Client-level field so the single-flight
// guard already spans sessions. The download is only ever aborted by the
// caller of Run cancelling the agent's whole lifetime; short of that, it
// runs to completion (or its own HTTP/sha error) in the background.
//
// A successful upgrade re-execs the process, abandoning any reconcile step
// mid-flight on this session's worker goroutine — safe because VMs are
// external processes unaffected by the agent's own exit, and engine state
// writes are atomic temp-and-rename, so whatever the abandoned step hadn't
// finished is simply re-derived, level-triggered, once the new binary
// reconnects and reads the next snapshot.
func (c *Client) maybeUpgrade(ctx context.Context, up *pb.AgentUpgrade) {
if up.GetVersion() == "" || up.GetVersion() == version.Version {
return
}
if !c.upgrading.CompareAndSwap(false, true) {
return
}
apply := c.applyUpgrade
if apply == nil {
apply = (&selfupdate.Applier{}).Apply
}
go func() {
slog.Info("agent self-upgrade starting", "from", version.Version, "to", up.GetVersion(), "url", up.GetUrl())
if err := apply(ctx, selfupdate.Update{Version: up.GetVersion(), URL: up.GetUrl(), SHA256: up.GetSha256()}); err != nil {
slog.Error("agent self-upgrade failed; will retry on next snapshot", "to", up.GetVersion(), "err", err)
c.upgrading.Store(false)
}
}()
}
// convergeExposures drives this host's listeners toward the snapshot and
// returns the rows the report carries. Level-triggered like everything else:
// every snapshot re-converges, so a bind that lost its port to a squatting
// process is retried on the next tick.
func (c *Client) convergeExposures(snap *pb.Snapshot) []*pb.ExposureStatus {
if c.Exposures == nil {
return nil
}
return c.Exposures.Converge(snap.GetExposures())
}
// reportExposures converges only against a snapshot the engine ACCEPTED. A
// fenced snapshot is one this host has already moved past, and driving
// listeners from it would re-open a port the fleet has since revoked — the
// same reason the fence path touches nothing else.
//
// There is a third state, and it deliberately DOES converge: a snapshot the
// engine refused for naming a min_agent_version above this build. Such a
// snapshot is current — it is not fenced — and an exposure spec is made of
// fields this agent already reads in full, so the floor tells us nothing about
// them. Tearing down live port-forwards because a NEWER field elsewhere in the
// snapshot is unreadable would take working published ports away from a fleet
// whose only fault is being one release behind, and help nobody. The engine
// signals this by returning FenceViolation false, which is what the check
// below keys on.
func (c *Client) reportExposures(snap *pb.Snapshot, rep *pb.Report) []*pb.ExposureStatus {
if rep.GetFenceViolation() {
return nil
}
return c.convergeExposures(snap)
}
// errPermanentAuth marks a credential rejection so Run() backs off long instead
// of tight-looping a dead credential.
var errPermanentAuth = errors.New("auth rejected (permanent)")
// Run loops forever: open a session, and on error back off then retry.
// Exits when ctx is cancelled.
func (c *Client) Run(ctx context.Context) {
failures := 0
for {
err := c.session(ctx)
if errors.Is(err, context.Canceled) {
return
}
if errors.Is(err, errPermanentAuth) {
slog.Error("not retrying quickly: credential rejected")
if !sleep(ctx, 60*time.Second) {
return
}
continue
}
// A cert pin mismatch (from transport.ClientTLS's VerifyConnection) is not
// a network-reachability problem, so it must not increment the UDP-blocked
// counter or fire that misleading warning. Treat it like the permanent path:
// log a distinct, actionable ERROR and back off long.
if err != nil && strings.Contains(err.Error(), "fingerprint mismatch") {
slog.Error("server cert pin mismatch — the server cert changed or enrollment is stale; re-enroll this host")
if !sleep(ctx, 60*time.Second) {
return
}
continue
}
// A session that successfully read ≥1 snapshot returns errSessionConnected
// wrapped around the underlying error; reset the failure counter so a
// long-lived-then-dropped connection is not mistaken for an unreachable
// control plane. (Documented choice: the connectedOnce sentinel keeps the
// UDP-blocked diagnostic accurate — it only fires when no handshake ever
// succeeded across N attempts.)
if errors.Is(err, errSessionConnected) {
failures = 0
} else {
failures++
if failures >= 3 {
slog.Warn("control-plane sync failing repeatedly — server unreachable OR wedged (accepting connections but not serving snapshots); check the server before suspecting the network",
"server", c.Identity.ServerQUICAddr, "consecutive_failures", failures, "last_err", err)
}
}
backoff := c.ReconnectBackoff
if backoff == 0 {
backoff = 5 * time.Second
}
slog.Warn("sync session ended, retrying", "err", err, "backoff", backoff)
if !sleep(ctx, backoff) {
return
}
}
}
func sleep(ctx context.Context, d time.Duration) bool {
select {
case <-ctx.Done():
return false
case <-time.After(d):
return true
}
}
// errSessionConnected is wrapped onto a session's terminal error once that
// session has read at least one snapshot, so Run() can reset its failure count.
var errSessionConnected = errors.New("session was connected")
// helloFacts gathers host facts and stamps the agent's own version onto them —
// the one fact the agent knows about itself rather than the host.
func helloFacts(ctx context.Context, run agentexec.Runner) *pb.HostFacts {
f := hostinfo.Facts(ctx, run)
f.AgentVersion = version.Version
return f
}
// session opens one QUIC connection, runs the dual-stream loop, and returns
// when the session ends (for any reason). The caller retries.
func (c *Client) session(ctx context.Context) error {
tlsConf := transport.ClientTLS(c.Identity.ServerCertSHA256)
conn, err := quic.DialAddr(ctx, c.Identity.ServerQUICAddr, tlsConf,
// Shared with the server listener via transport so the two ends can't drift.
transport.SyncQUICConfig())
if err != nil {
return classifyErr(err) // transient: Run() backs off
}
defer conn.CloseWithError(0, "")
up, err := conn.OpenStreamSync(ctx)
if err != nil {
return classifyErr(err)
}
stateDir := c.StateDir
if stateDir == "" {
stateDir = "/var/lib/eitri-agent"
}
hostname, _ := os.Hostname()
hello := &pb.AgentMessage{Msg: &pb.AgentMessage_Hello{Hello: &pb.Hello{
HostId: c.Identity.HostID, Hostname: hostname, Os: runtime.GOOS, Arch: runtime.GOARCH,
Provisioner: c.Provisioner,
LastSeenEpoch: c.St.Epoch(), Capacity: c.advertisedCapacity(stateDir),
Facts: helloFacts(ctx, c.Runner),
HostNetworks: c.HostNetworks,
Credential: c.Identity.Credential,
}}}
if err := transport.WriteMsg(up, hello); err != nil {
return classifyErr(err)
}
// Accept the server's down-stream (visible on first server write).
down, err := conn.AcceptStream(ctx)
if err != nil {
// If the server rejected auth, CloseWithError surfaces here as an
// ApplicationError with CodeAuthRejected.
return classifyErr(err)
}
// Console streams: every server-initiated stream after the snapshot
// down-stream is a console request. The loop dies with the connection,
// which also ends every console session on it — the browser reconnects.
go func() {
for {
cs, err := conn.AcceptStream(ctx)
if err != nil {
return
}
go c.handleConsoleStream(ctx, cs)
}
}()
var mu sync.Mutex
var latest *pb.Snapshot
// connectedOnce is set true (under mu) once the recv goroutine reads its
// first snapshot. session wraps its terminal error with errSessionConnected
// when set, so Run() resets its consecutive-failure counter.
connectedOnce := false
stepSignal := make(chan struct{}, 1)
errc := make(chan error, 2)
// sessCtx is cancelled when session returns (for ANY reason), giving every
// worker goroutine a signal to exit. Most sessions end because the recv
// goroutine errored — NOT because the parent ctx was cancelled — so a worker
// that selects only on the long-lived parent ctx (plus a dead stepSignal and
// a stopped ticker) would never observe session end and would leak one
// goroutine per reconnect.
sessCtx, cancel := context.WithCancel(ctx)
defer cancel()
// Remembered across ticks so the subnet is logged when it CHANGES rather
// than every ten seconds. A host that cannot say which network its guests
// are on is otherwise indistinguishable from one that can: the empty answer
// is correct on the wire — the fleet keeps its existing record — and so is
// silent everywhere an operator might look.
lastCIDR := unreported
step := func() error {
mu.Lock()
snap := latest
mu.Unlock()
if snap == nil {
return nil
}
rep := c.Engine.Step(ctx, snap)
rep.Capacity = c.advertisedCapacity(stateDir)
rep.Metrics = hostinfo.Metrics(stateDir)
rep.Exposures = c.reportExposures(snap, rep)
// Polled, not resolved once: a host's address can change under a
// live agent (a lease renews on a different address). Empty is "no
// answer" and leaves the fleet's record alone.
if c.UplinkAddr != nil {
rep.HostUplinkAddr = c.UplinkAddr()
}
// Polled, not resolved once: a host's guest subnet can change while the
// agent stays connected, and on a platform whose OS owns the network it
// may not be knowable at connect time at all. Empty is "no answer" and
// leaves the fleet's record alone.
if c.GuestCIDR != nil {
rep.GuestCidr = c.GuestCIDR()
logGuestCIDR(rep.GuestCidr, &lastCIDR)
}
return transport.WriteMsg(up, &pb.AgentMessage{Msg: &pb.AgentMessage_Report{Report: rep}})
}
// Recv goroutine: read down-stream snapshots only; never writes.
go func() {
for {
var msg pb.ServerMessage
if err := transport.ReadMsg(down, &msg, transport.DefaultMaxFrame); err != nil {
errc <- classifyErr(err)
return
}
if snap := msg.GetSnapshot(); snap != nil {
mu.Lock()
latest = snap
connectedOnce = true
mu.Unlock()
select {
case stepSignal <- struct{}{}:
default:
}
if up := snap.GetAgentUpgrade(); up != nil {
c.maybeUpgrade(ctx, up)
}
}
}
}()
// Worker goroutine: owns all up-stream writes (snapshot-driven + ticker).
tick := c.TickInterval
if tick == 0 {
tick = DefaultTickInterval
}
ticker := time.NewTicker(tick)
defer ticker.Stop()
go func() {
for {
select {
case <-sessCtx.Done():
errc <- sessCtx.Err()
return
case <-stepSignal:
if err := step(); err != nil {
errc <- err
return
}
case <-ticker.C:
if err := step(); err != nil {
errc <- err
return
}
}
}
}()
sessErr := <-errc
mu.Lock()
connected := connectedOnce
mu.Unlock()
if connected && !errors.Is(sessErr, errPermanentAuth) {
// Preserve errPermanentAuth's semantics; otherwise mark the session as
// having connected so Run() resets its failure counter.
return errors.Join(errSessionConnected, sessErr)
}
return sessErr
}
// handleConsoleStream services one server-opened console stream: read the
// ConsoleOpen header, reply ConsoleOpened (ok=false with a reason when the VM
// has no pump or no Console is wired), then hand the raw stream to the pump.
func (c *Client) handleConsoleStream(ctx context.Context, s quic.Stream) {
defer func() {
s.CancelRead(0)
_ = s.Close()
}()
// Defense-in-depth: bound the header read so a stream that never delivers
// its first frame can't pin this goroutine forever. Only a server bug can
// trip it — the server writes ConsoleOpen immediately or closes — so 10s
// (mirroring the server's consoleHandshakeTimeout) is generous.
_ = s.SetReadDeadline(time.Now().Add(10 * time.Second))
var first pb.ServerMessage
if err := transport.ReadMsg(s, &first, transport.DefaultMaxFrame); err != nil {
return
}
_ = s.SetReadDeadline(time.Time{})
if tcp := first.GetTcpOpen(); tcp != nil {
// Raw TCP tunnel (e.g. the SSH jump gate), not a console. tunnelStream's
// Close tears down both directions so either copy goroutine can unblock
// the other; the deferred Close above is then a harmless second close.
c.handleTCPStream(ctx, tunnelStream{s}, tcp.GetVmId(), tcp.GetPort())
return
}
open := first.GetConsoleOpen()
if open == nil {
return // not a console stream; drop it
}
refuse := func(msg string) {
_ = transport.WriteMsg(s, &pb.AgentMessage{Msg: &pb.AgentMessage_ConsoleOpened{
ConsoleOpened: &pb.ConsoleOpened{Ok: false, Error: msg}}})
}
if c.Console == nil {
refuse("console not supported by this agent")
return
}
sentReady := false
err := c.Console.Attach(ctx, open.GetVmId(), s, func() error {
if err := transport.WriteMsg(s, &pb.AgentMessage{Msg: &pb.AgentMessage_ConsoleOpened{
ConsoleOpened: &pb.ConsoleOpened{Ok: true}}}); err != nil {
return err
}
sentReady = true
return nil
})
if err != nil {
if !sentReady {
// Validation failed before the ok-frame: an ok=false reply is
// still legal wire protocol.
refuse(err.Error())
}
// After sentReady the stream is RAW bytes: writing a ConsoleOpened
// frame here would inject protobuf garbage into a LIVE console (the
// peer is alive on a slow-viewer drop or VM-deleted-mid-session) —
// late errors just close the stream (the deferred Close above).
slog.Debug("console session ended", "vm", open.GetVmId(), "err", err)
}
}
// tunnelStream adapts a quic.Stream so Close tears down BOTH directions
// (Stream.Close closes only the write half). handleTCPStream needs a closing a
// blocked Read: cancelling the read half is how one copy goroutine unblocks the
// other when the peer TCP conn dies.
type tunnelStream struct{ quic.Stream }
func (t tunnelStream) Close() error {
t.CancelRead(0)
return t.Stream.Close()
}
// handleTCPStream services a server-opened raw TCP tunnel: it validates the
// request against agent state, replies TCPOpened, then splices the stream to a
// fresh dial of the VM's guest ssh port. It deliberately does NOT go through
// serialpump — the pump's replay ring and slow-consumer drop would corrupt an
// SSH byte stream. On any refusal it sends ok=false and returns.
func (c *Client) handleTCPStream(ctx context.Context, stream io.ReadWriteCloser, vmID string, port uint32) {
refuse := func(msg string) {
_ = transport.WriteMsg(stream, &pb.AgentMessage{Msg: &pb.AgentMessage_TcpOpened{
TcpOpened: &pb.TCPOpened{Ok: false, Error: msg}}})
}
rec, ok, err := c.St.Get(vmID)
if err != nil {
// The record is there but unreadable, so whether this host runs the VM
// cannot be answered — refuse without claiming it is absent.
refuse("vm record unreadable on this host")
return
}
if !ok {
refuse("vm not on this host")
return
}
if rec.IP == "" {
// An empty IP would make JoinHostPort produce ":22", so net.Dial would
// hit the AGENT host's own sshd — refuse instead of tunnelling to self.
refuse("vm has no address")
return
}
if port != 22 {
// Pin 22; the wire port is validated but never trusted for the dial.
refuse("port not allowed")
return
}
conn, err := c.dial(rec.IP)
if err != nil {
refuse(err.Error())
return
}
defer conn.Close()
if err := transport.WriteMsg(stream, &pb.AgentMessage{Msg: &pb.AgentMessage_TcpOpened{
TcpOpened: &pb.TCPOpened{Ok: true}}}); err != nil {
return
}
// Fresh dual splice: each direction, on EOF/error, closes both ends so the
// opposite goroutine unblocks and exits. Both ends are closed on return.
done := make(chan struct{}, 2)
go func() {
_, _ = io.Copy(conn, stream)
conn.Close()
stream.Close()
done <- struct{}{}
}()
go func() {
_, _ = io.Copy(stream, conn)
conn.Close()
stream.Close()
done <- struct{}{}
}()
<-done
<-done
}
// dial connects to a VM's ssh port. Production pins port 22; tests override
// via c.dialGuest.
func (c *Client) dial(ip string) (net.Conn, error) {
if c.dialGuest != nil {
return c.dialGuest(ip)
}
return net.Dial("tcp", net.JoinHostPort(ip, "22"))
}
// unreported distinguishes "no report has been sent yet" from a report that
// carried no subnet, so the first answer is always logged — including the first
// empty one.
const unreported = "\x00never reported"
// logGuestCIDR reports a change in this host's guest subnet, once, at the tick
// it changes. An answer that goes away is a WARNING: the fleet will keep
// answering from its existing record, which is right, and quietly wrong-looking
// if the host's network has genuinely moved.
func logGuestCIDR(cidr string, last *string) {
if cidr == *last {
return
}
*last = cidr
if cidr == "" {
slog.Warn("guest subnet unknown — the fleet keeps this host's existing record")
return
}
slog.Info("guest subnet", "cidr", cidr)
}
// classifyErr distinguishes a permanent auth rejection from a transient error so
// Run() can log loudly and avoid a tight reconnect loop on a dead credential.
func classifyErr(err error) error {
var appErr *quic.ApplicationError
if !errors.As(err, &appErr) {
return err
}
switch appErr.ErrorCode {
case transport.CodeAuthRejected:
slog.Error("host credential rejected by server — re-enroll this host", "detail", appErr.ErrorMessage)
return errPermanentAuth
case transport.CodeSuperseded:
// Transient on purpose: this agent reconnects on the normal backoff. Said
// loudly because the healthy cause (this agent was restarted and the
// server still held the old session) is indistinguishable on the wire
// from the unhealthy one — a second agent elsewhere carrying a copy of
// this host's identity, which leaves the two evicting each other forever.
slog.Warn("another agent connected as this host and took over the session; reconnecting — if this repeats, two agents are sharing one identity")
}
return err
}