a73x

d518b6b2

fix(sync): one agent per identity — a second launch is refused, a second session evicted

a73x   2026-08-10 15:30

Commit message
fix(sync): one agent per identity — a second launch is refused, a second session evicted

An agent's identity is its state directory, and two agents sharing one are
not two agents: they are one host reconciling its own VMs twice from a
half-observed view of the disk, and connecting to the control plane twice
as the same host. It happens by accident — a second launch that forgets
to repeat --state-dir and lands back on the default.

The control plane kept only the newer of the two sessions and left the
older one hanging. The elder's handler stopped reading the moment its
poke channel was displaced, but nothing closed the connection, so the
displaced agent kept writing reports into a stream with no reader and
QUIC keepalives held the socket open indefinitely. It never redialled,
because from where it stood nothing had gone wrong. When the newer
session ended, the host's last contact stayed frozen at whatever the
elder had last managed to say, and the fleet showed it offline until
someone broke that connection by hand.

So the newest Hello for a host wins and the session it displaces is
closed, with a code that says superseded rather than rejected: the
displaced agent sees the close, comes straight back on its ordinary
five-second backoff, and one of the two ends up holding the session
instead of both ending up mute. An agent told this twice says so, since
the healthy cause — it was simply restarted — looks identical on the wire
to two agents elsewhere sharing one identity.

And no session may end any other way now. Every exit from the sync
handler closes its connection, so a reader that goes away can no longer
leave a live transport behind it.

The refusal is the other half. An agent takes an exclusive lock on its
state directory before it touches the network, and a second one that
finds the lock held exits naming the process that holds it, rather than
starting up to fight over the same records. The lock is the directory's,
not the machine's: a host may run several agents, each enrolled
separately, each with its own state. It is dropped on exec, which is what
lets an agent that replaces its own binary re-take the lock it held a
moment earlier.
docs/shape.html
Old New
@@ -248,6 +248,7 @@
248 "internal/agent/seed", 248 "internal/agent/seed",
249 "internal/agent/serialpump", 249 "internal/agent/serialpump",
250 "internal/agent/state", 250 "internal/agent/state",
251 "internal/agent/statelock",
251 "internal/agent/syncclient", 252 "internal/agent/syncclient",
252 "internal/covsnap", 253 "internal/covsnap",
253 "internal/joinblob" 254 "internal/joinblob"
@@ -278,6 +279,12 @@
278 "imports": [] 279 "imports": []
279 }, 280 },
280 { 281 {
282 "importPath": "internal/agent/statelock",
283 "plane": "data",
284 "synopsis": "Package statelock enforces one running agent per identity.",
285 "imports": []
286 },
287 {
281 "importPath": "internal/agent/syncclient", 288 "importPath": "internal/agent/syncclient",
282 "plane": "data", 289 "plane": "data",
283 "synopsis": "Package syncclient holds the agent's stream loop: receive snapshots, run engine steps, send reports.", 290 "synopsis": "Package syncclient holds the agent's stream loop: receive snapshots, run engine steps, send reports.",
docs/shape.json
Old New
@@ -197,6 +197,7 @@
197 "internal/agent/seed", 197 "internal/agent/seed",
198 "internal/agent/serialpump", 198 "internal/agent/serialpump",
199 "internal/agent/state", 199 "internal/agent/state",
200 "internal/agent/statelock",
200 "internal/agent/syncclient", 201 "internal/agent/syncclient",
201 "internal/covsnap", 202 "internal/covsnap",
202 "internal/joinblob" 203 "internal/joinblob"
@@ -227,6 +228,12 @@
227 "imports": [] 228 "imports": []
228 }, 229 },
229 { 230 {
231 "importPath": "internal/agent/statelock",
232 "plane": "data",
233 "synopsis": "Package statelock enforces one running agent per identity.",
234 "imports": []
235 },
236 {
230 "importPath": "internal/agent/syncclient", 237 "importPath": "internal/agent/syncclient",
231 "plane": "data", 238 "plane": "data",
232 "synopsis": "Package syncclient holds the agent's stream loop: receive snapshots, run engine steps, send reports.", 239 "synopsis": "Package syncclient holds the agent's stream loop: receive snapshots, run engine steps, send reports.",
internal/agent/run/cli.go
Old New
@@ -29,6 +29,7 @@ import (
29 "github.com/a73x/eitri/internal/agent/reconcile" 29 "github.com/a73x/eitri/internal/agent/reconcile"
30 "github.com/a73x/eitri/internal/agent/seed" 30 "github.com/a73x/eitri/internal/agent/seed"
31 "github.com/a73x/eitri/internal/agent/state" 31 "github.com/a73x/eitri/internal/agent/state"
32 "github.com/a73x/eitri/internal/agent/statelock"
32 "github.com/a73x/eitri/internal/agent/syncclient" 33 "github.com/a73x/eitri/internal/agent/syncclient"
33 "github.com/a73x/eitri/internal/covsnap" 34 "github.com/a73x/eitri/internal/covsnap"
34 "github.com/a73x/eitri/internal/joinblob" 35 "github.com/a73x/eitri/internal/joinblob"
@@ -213,6 +214,17 @@ func join(st *state.Store, cfg Config, blob string) error {
213 // serve handles the normal (no subcommand) run mode: it wires the reconcile 214 // serve handles the normal (no subcommand) run mode: it wires the reconcile
214 // engine and sync client and blocks until SIGINT/SIGTERM. 215 // engine and sync client and blocks until SIGINT/SIGTERM.
215 func serve(st *state.Store, cfg Config) error { 216 func serve(st *state.Store, cfg Config) error {
217 // First, before this agent touches the network, a bridge, or a VM record:
218 // claim the state directory. Everything below assumes this process is the
219 // only one driving this identity — the VM records it reconciles, the epoch
220 // fence it advances, and the one sync session the control plane keeps per
221 // host. Held until the process exits.
222 lk, err := statelock.Acquire(cfg.StateDir)
223 if err != nil {
224 return err
225 }
226 defer func() { _ = lk.Release() }()
227
216 id, ok := st.Identity() 228 id, ok := st.Identity()
217 if !ok { 229 if !ok {
218 return errors.New("not enrolled — run with 'join <blob>' subcommand first") 230 return errors.New("not enrolled — run with 'join <blob>' subcommand first")
internal/agent/statelock/statelock.go
Old New
@@ -0,0 +1,102 @@
1 // Package statelock enforces one running agent per identity.
2 //
3 // An agent's identity is its state directory: identity.json, the host
4 // credential in it, the epoch fence, and every VM record. Two agents sharing
5 // one of those are not two agents — they are one host answering the control
6 // plane twice, each reconciling the other's VMs from a half-observed view of
7 // the disk, and each connecting as the same host so that only one of them is
8 // ever heard. It happens by accident, most often by launching a second agent
9 // without repeating --state-dir and landing back on the default.
10 //
11 // So the state directory is the thing locked, and deliberately not the machine:
12 // one host may legitimately run several agents, each with its own directory and
13 // its own identity in the fleet. Locking anything global would forbid that.
14 package statelock
15
16 import (
17 "errors"
18 "fmt"
19 "os"
20 "path/filepath"
21 "strconv"
22 "strings"
23
24 "golang.org/x/sys/unix"
25 )
26
27 // lockName is the file inside the state directory whose flock IS the claim. It
28 // carries the holder's pid as its contents, which is diagnostic only — the lock
29 // itself is what the kernel enforces, so a stale pid can never wrongly refuse
30 // anyone.
31 const lockName = "agent.lock"
32
33 // mode matches every other file in the state directory: the agent's alone.
34 const mode = 0o600
35
36 // Lock is a held claim on one state directory. It owns an open file whose flock
37 // lives for as long as the file stays open, so the value must outlive every
38 // caller that relies on the claim.
39 type Lock struct{ f *os.File }
40
41 // Acquire claims dir for this process, and fails rather than waits when another
42 // agent already holds it — a second launch is a mistake to report, not a queue
43 // to join.
44 //
45 // The lock is released by exec, on purpose. flock ownership rides the open file
46 // description, and the agent replaces its own binary in place: it renames the
47 // new image over itself and re-execs, keeping its pid (KillMode=process in the
48 // unit exists to let that happen). An fd that survived exec would carry the
49 // claim into the new image, which would then be unable to take a lock it
50 // already holds — and could not tell that deadlock apart from a real second
51 // agent. Every fd Go opens is close-on-exec, which is what makes the re-exec'd
52 // binary land on a free lock and re-take it immediately; the gap is the width
53 // of an execve, with no other launcher racing for it. The flag is spelled out
54 // here so that this stays true if the open is ever rewritten.
55 func Acquire(dir string) (*Lock, error) {
56 path := filepath.Join(dir, lockName)
57 f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|unix.O_CLOEXEC, mode)
58 if err != nil {
59 return nil, fmt.Errorf("open agent lock: %w", err)
60 }
61 if err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
62 held := holder(path)
63 f.Close()
64 if errors.Is(err, unix.EWOULDBLOCK) {
65 if held <= 0 {
66 return nil, errors.New("an agent is already running with this identity")
67 }
68 return nil, fmt.Errorf("an agent is already running with this identity (pid %d)", held)
69 }
70 return nil, fmt.Errorf("lock agent state dir: %w", err)
71 }
72 // Ours now — say whose. Truncate first: a shorter pid than the last holder's
73 // would otherwise leave that one's trailing digits behind.
74 if err := f.Truncate(0); err != nil {
75 f.Close()
76 return nil, fmt.Errorf("truncate agent lock: %w", err)
77 }
78 if _, err := f.WriteAt([]byte(strconv.Itoa(os.Getpid())+"\n"), 0); err != nil {
79 f.Close()
80 return nil, fmt.Errorf("write agent lock: %w", err)
81 }
82 return &Lock{f: f}, nil
83 }
84
85 // Release drops the claim. The agent calls it only on its way out — the lock is
86 // held for the whole run, not around any particular piece of work.
87 func (l *Lock) Release() error { return l.f.Close() }
88
89 // holder reads the pid the current holder wrote, or 0 when the file says
90 // nothing usable. Only ever used to make the refusal name a process; the
91 // refusal itself stands on the kernel's answer.
92 func holder(path string) int {
93 raw, err := os.ReadFile(path)
94 if err != nil {
95 return 0
96 }
97 pid, err := strconv.Atoi(strings.TrimSpace(string(raw)))
98 if err != nil {
99 return 0
100 }
101 return pid
102 }
internal/agent/statelock/statelock_test.go
Old New
@@ -0,0 +1,86 @@
1 package statelock
2
3 import (
4 "fmt"
5 "os"
6 "path/filepath"
7 "testing"
8
9 "github.com/stretchr/testify/assert"
10 "github.com/stretchr/testify/require"
11 )
12
13 // flock ownership belongs to the open file description, not the process, so two
14 // Acquires in this one test binary conflict exactly as two agents would.
15
16 func TestAcquireClaimsTheDirectoryAndNamesTheHolder(t *testing.T) {
17 dir := t.TempDir()
18
19 lk, err := Acquire(dir)
20 require.NoError(t, err)
21 t.Cleanup(func() { lk.Release() }) //nolint:errcheck
22
23 raw, err := os.ReadFile(filepath.Join(dir, lockName))
24 require.NoError(t, err)
25 assert.Equal(t, fmt.Sprintf("%d\n", os.Getpid()), string(raw),
26 "the lock file names the process holding it")
27 }
28
29 func TestASecondAgentIsRefusedAndToldWhichProcessHasIt(t *testing.T) {
30 dir := t.TempDir()
31 lk, err := Acquire(dir)
32 require.NoError(t, err)
33 t.Cleanup(func() { lk.Release() }) //nolint:errcheck
34
35 second, err := Acquire(dir)
36 assert.Nil(t, second)
37 require.Error(t, err)
38 assert.Equal(t, fmt.Sprintf("an agent is already running with this identity (pid %d)", os.Getpid()), err.Error())
39 }
40
41 func TestARefusalWithNoReadablePidStillRefuses(t *testing.T) {
42 dir := t.TempDir()
43 lk, err := Acquire(dir)
44 require.NoError(t, err)
45 t.Cleanup(func() { lk.Release() }) //nolint:errcheck
46
47 // An agent old enough to predate the pid, or one killed between taking the
48 // lock and writing it. The kernel still says the lock is held, which is the
49 // answer that counts; the message simply names nobody.
50 require.NoError(t, os.WriteFile(filepath.Join(dir, lockName), []byte("not a pid\n"), mode))
51
52 _, err = Acquire(dir)
53 require.Error(t, err)
54 assert.Equal(t, "an agent is already running with this identity", err.Error())
55 }
56
57 func TestReleasingLetsTheNextAgentIn(t *testing.T) {
58 dir := t.TempDir()
59 lk, err := Acquire(dir)
60 require.NoError(t, err)
61 require.NoError(t, lk.Release())
62
63 // An agent that has exited leaves nothing behind but a file: a restart, or
64 // the re-exec of a self-upgrade, must not be refused by its own predecessor.
65 next, err := Acquire(dir)
66 require.NoError(t, err)
67 assert.NoError(t, next.Release())
68 }
69
70 func TestTwoStateDirsAreTwoIdentities(t *testing.T) {
71 // One machine may legitimately run several agents, each enrolled separately.
72 // The lock is per state directory precisely so that stays possible.
73 a, err := Acquire(t.TempDir())
74 require.NoError(t, err)
75 t.Cleanup(func() { a.Release() }) //nolint:errcheck
76
77 b, err := Acquire(t.TempDir())
78 require.NoError(t, err)
79 assert.NoError(t, b.Release())
80 }
81
82 func TestAcquireFailsWhenTheDirectoryIsNotThere(t *testing.T) {
83 _, err := Acquire(filepath.Join(t.TempDir(), "no-such-dir"))
84 require.Error(t, err)
85 assert.Contains(t, err.Error(), "open agent lock")
86 }
internal/agent/syncclient/client.go
Old New
@@ -646,9 +646,20 @@ func logGuestCIDR(cidr string, last *string) {
646 // Run() can log loudly and avoid a tight reconnect loop on a dead credential. 646 // Run() can log loudly and avoid a tight reconnect loop on a dead credential.
647 func classifyErr(err error) error { 647 func classifyErr(err error) error {
648 var appErr *quic.ApplicationError 648 var appErr *quic.ApplicationError
649 if errors.As(err, &appErr) && appErr.ErrorCode == transport.CodeAuthRejected { 649 if !errors.As(err, &appErr) {
650 return err
651 }
652 switch appErr.ErrorCode {
653 case transport.CodeAuthRejected:
650 slog.Error("host credential rejected by server — re-enroll this host", "detail", appErr.ErrorMessage) 654 slog.Error("host credential rejected by server — re-enroll this host", "detail", appErr.ErrorMessage)
651 return errPermanentAuth 655 return errPermanentAuth
656 case transport.CodeSuperseded:
657 // Transient on purpose: this agent reconnects on the normal backoff. Said
658 // loudly because the healthy cause (this agent was restarted and the
659 // server still held the old session) is indistinguishable on the wire
660 // from the unhealthy one — a second agent elsewhere carrying a copy of
661 // this host's identity, which leaves the two evicting each other forever.
662 slog.Warn("another agent connected as this host and took over the session; reconnecting — if this repeats, two agents are sharing one identity")
652 } 663 }
653 return err 664 return err
654 } 665 }
internal/server/syncsvc/syncsvc.go
Old New
@@ -130,6 +130,14 @@ var helloGrace = 30 * time.Second
130 // NOTE (verified Task 1): quic-go v0.48.2 uses INTERFACES quic.Connection and 130 // NOTE (verified Task 1): quic-go v0.48.2 uses INTERFACES quic.Connection and
131 // quic.Stream (not *quic.Conn/*quic.Stream, which only exist in v0.49+). 131 // quic.Stream (not *quic.Conn/*quic.Stream, which only exist in v0.49+).
132 func (s *Service) handleConn(ctx context.Context, conn quic.Connection) { 132 func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
133 // Every way out of here closes the connection. A handler that returns while
134 // leaving the transport up leaves an agent writing reports nobody reads —
135 // and QUIC keepalives hold that socket open indefinitely, so the host's last
136 // contact freezes and it reads offline until someone breaks the connection by
137 // hand. quic-go closes once (closeOnce), so the specific codes the auth paths
138 // below send still reach the agent; this only catches whatever they miss.
139 defer conn.CloseWithError(0, "session ended")
140
133 // Up-stream: agent opens it and sends Hello first. Both the accept and the 141 // Up-stream: agent opens it and sends Hello first. Both the accept and the
134 // first read run under the hello grace so an unauthenticated peer that 142 // first read run under the hello grace so an unauthenticated peer that
135 // handshakes then stalls cannot park this goroutine indefinitely. 143 // handshakes then stalls cannot park this goroutine indefinitely.
@@ -203,9 +211,24 @@ func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
203 // Console reachability: only now — with the down-stream open — is it safe 211 // Console reachability: only now — with the down-stream open — is it safe
204 // for OpenConsole to add streams to this connection (stream-order 212 // for OpenConsole to add streams to this connection (stream-order
205 // invariant: the snapshot down-stream is always the first accepted). 213 // invariant: the snapshot down-stream is always the first accepted).
214 //
215 // One host, one session: a second Hello for a host already connected wins,
216 // and the session it displaces is closed rather than abandoned. Two agents
217 // sharing an identity are a misconfiguration, and the newest connection is
218 // the one that just proved it holds the credential — but the displaced side
219 // must SEE the close, or it sits there reporting into a stream that has no
220 // reader. CodeSuperseded puts it on the normal reconnect backoff, so a host
221 // whose agent was merely restarted comes straight back.
206 s.consoleMu.Lock() 222 s.consoleMu.Lock()
223 displaced := s.conns[hostID]
207 s.conns[hostID] = conn 224 s.conns[hostID] = conn
208 s.consoleMu.Unlock() 225 s.consoleMu.Unlock()
226 if displaced != nil {
227 // Outside the lock: CloseWithError waits for the displaced connection to
228 // finish tearing down, and the console broker must not queue behind that.
229 slog.Info("newer session for this host; closing the older one", "host", hostID)
230 _ = displaced.CloseWithError(transport.CodeSuperseded, "superseded by a newer session for this host")
231 }
209 defer func() { 232 defer func() {
210 s.consoleMu.Lock() 233 s.consoleMu.Lock()
211 // Only deregister OUR conn: a reconnect may already have replaced it. 234 // Only deregister OUR conn: a reconnect may already have replaced it.
internal/server/syncsvc/syncsvc_test.go
Old New
@@ -659,10 +659,11 @@ func TestReconnectKeepsNewConnRegistered(t *testing.T) {
659 } 659 }
660 }() 660 }()
661 661
662 // Close A. Its handler's deferred deregistration runs asynchronously; the 662 // A has already been evicted by B's arrival; closing it from this end too
663 // guard (only delete if the map still holds OUR conn) must leave B alone — 663 // changes nothing and is what a real displaced agent does next. Either way
664 // so across the whole teardown window no attempt may ever see 664 // A's handler deregisters asynchronously, and the guard (only delete if the
665 // ErrAgentOffline. 665 // map still holds OUR conn) must leave B alone — so across the whole
666 // teardown window no attempt may ever see ErrAgentOffline.
666 a.conn.CloseWithError(0, "") 667 a.conn.CloseWithError(0, "")
667 require.Never(t, func() bool { 668 require.Never(t, func() bool {
668 stream, err := f.svc.OpenConsole(context.Background(), f.host.ID, "vm-1") 669 stream, err := f.svc.OpenConsole(context.Background(), f.host.ID, "vm-1")
@@ -905,3 +906,47 @@ func TestReportRecordsTheHostsGuestSubnet(t *testing.T) {
905 return err == nil && h.BridgeCIDR != "192.168.64.0/24" 906 return err == nil && h.BridgeCIDR != "192.168.64.0/24"
906 }, 500*time.Millisecond, 25*time.Millisecond, "an empty report must leave the record alone") 907 }, 500*time.Millisecond, 25*time.Millisecond, "an empty report must leave the record alone")
907 } 908 }
909
910 // TestSecondSessionEvictsTheFirst pins one host to one session. Before it, the
911 // hub silently displaced the elder's poke channel and the elder's handler
912 // returned without closing the connection: the agent on the other end kept
913 // writing reports into a stream with no reader, keepalives held the socket open
914 // indefinitely, and the host's last-seen froze until someone broke the
915 // connection by hand. The elder must SEE the close, and see it as transient so
916 // a live agent comes straight back rather than sitting out the auth backoff.
917 func TestSecondSessionEvictsTheFirst(t *testing.T) {
918 f := setup(t)
919 elder := mustDial(t, f)
920 elder.recv(t) // elder is the registered session
921
922 newer := mustDial(t, f)
923 newer.recv(t) // newer has taken over
924
925 require.NoError(t, elder.down.SetReadDeadline(time.Now().Add(5*time.Second)))
926 var discard pb.ServerMessage
927 err := transport.ReadMsg(elder.down, &discard, transport.DefaultMaxFrame)
928 require.Error(t, err, "the displaced session must be closed, not abandoned")
929 var appErr *quic.ApplicationError
930 require.True(t, errors.As(err, &appErr), "expected *quic.ApplicationError, got %T: %v", err, err)
931 assert.Equal(t, quic.ApplicationErrorCode(transport.CodeSuperseded), appErr.ErrorCode,
932 "the displaced agent must be told it was superseded, not that its credential is bad")
933
934 // The elder's teardown runs asynchronously and must take nothing of the
935 // newer's with it. Its console registration survives the whole window —
936 // deregistration is guarded on the map still holding the elder's own conn.
937 require.Never(t, func() bool {
938 f.svc.consoleMu.Lock()
939 defer f.svc.consoleMu.Unlock()
940 return f.svc.conns[f.host.ID] == nil
941 }, time.Second, 50*time.Millisecond,
942 "the evicted session's teardown deregistered the surviving one")
943
944 // And so does its hub subscription: a poke still reaches it.
945 require.NoError(t, f.st.CreateVM(store.VM{ID: "vm1", HostID: f.host.ID, Name: "a",
946 ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
947 f.hub.Poke(f.host.ID)
948 require.NoError(t, newer.down.SetReadDeadline(time.Now().Add(5*time.Second)))
949 snap := newer.recv(t).GetSnapshot()
950 require.NotNil(t, snap, "the surviving session must still be poked")
951 require.Len(t, snap.Vms, 1)
952 }
internal/transport/tlsconf.go
Old New
@@ -25,6 +25,12 @@ const ALPN = "eitri-sync/1"
25 // the server dropped it. 25 // the server dropped it.
26 const ( 26 const (
27 CodeAuthRejected = 1 // permanent: bad/expired host credential — do not tight-loop 27 CodeAuthRejected = 1 // permanent: bad/expired host credential — do not tight-loop
28 // CodeSuperseded closes the older of two sessions claiming one host: the
29 // newest Hello wins and the displaced agent is told so, rather than being
30 // left holding a connection nobody reads. It is transient by design — the
31 // displaced agent reconnects on its normal backoff, and only the auth code
32 // above earns the long one.
33 CodeSuperseded = 2
28 ) 34 )
29 35
30 // Sync QUIC keepalive/idle timings. The agent dialer and the server listener 36 // Sync QUIC keepalive/idle timings. The agent dialer and the server listener
scripts/coverage.sh
Old New
@@ -18,6 +18,7 @@ cd "$(dirname "$0")/.."
18 declare -A FLOOR=( 18 declare -A FLOOR=(
19 [internal/agent/reconcile]=81 19 [internal/agent/reconcile]=81
20 [internal/agent/state]=50 20 [internal/agent/state]=50
21 [internal/agent/statelock]=75
21 [internal/agent/seed]=79 22 [internal/agent/seed]=79
22 [internal/agent/ipalloc]=83 23 [internal/agent/ipalloc]=83
23 [internal/agent/hostinfo]=95 24 [internal/agent/hostinfo]=95