| index 7b7a5f07d..fed039d8e 100644 |
| --- a/internal/caps/converge.go |
| +++ b/internal/caps/converge.go |
| @@ -65,10 +65,17 @@ type Plan struct { |
| // pass never wrote it, so putting it back is not this pass's business — and |
| // Restore is marker-checked too, so a "not had" half can never turn into the |
| // deletion of somebody's hand-written drop-in. |
| +// |
| +// Daemon travels with the snapshot because putting the BYTES back is only |
| +// half of putting the box back: a restored ListenStream that systemd never |
| +// re-binds restores nothing, and re-binding it needs the same stop → restart |
| +// → start ordering Converge uses, which needs to know whether the service is |
| +// supposed to be running. See Restore. |
| type Snapshot struct { |
| Host string |
| Service, Socket []byte |
| ServiceHad, SocketHad bool |
| + Daemon bool |
| } |
| |
| // Result is what Converge did. |
| @@ -110,10 +117,27 @@ func DropInPaths(etcDir, host string) (service, socket string) { |
| // - p.Set non-nil, renders bytes: write them if they differ. |
| // - p.Set non-nil, renders nil: REMOVE the generated file (marker-checked). |
| // |
| -// Every refusal returns an error with Changed=false and nothing written, |
| -// having already logged one line naming the operator's fix. Callers that |
| -// merely survey the box (the hourly grant-drift pass) can log-and-continue; |
| -// callers on the version-change path let it refuse the version. |
| +// Two different kinds of error come back, and they are not the same claim: |
| +// |
| +// - a REFUSAL (every check above, plus the claims and state-dir guards) is |
| +// Changed=false with nothing written, having already logged one line |
| +// naming the operator's fix. Callers that merely survey the box (the |
| +// hourly grant-drift pass) can log-and-continue; callers on the |
| +// version-change path let it refuse the version. |
| +// - a mid-pass IO or systemd FAILURE (a write, a daemon-reload, the socket |
| +// sequence) rolls the bytes back to Prev so that "the bytes on disk say |
| +// converged" stays a sound invariant — if systemd never loaded them, they |
| +// must not be on disk claiming it did, or the next pass would find no |
| +// diff and never retry. Changed is then false, because the disk is back |
| +// where it started. It is true only when even the roll-back failed, which |
| +// is the honest answer: bytes moved and this pass could not undo it. |
| +// |
| +// Failures AFTER a successful reload — an `enable`, a `start`, a daemon's |
| +// `restart` — deliberately do NOT roll the bytes back. systemd has the new |
| +// configuration and it is correct; the failure is the app's. Rolling back |
| +// there would flap the unit text between two shapes on every pass, and the |
| +// version-change caller already owns that case (it rolls back the symlink and |
| +// calls Restore). |
| // |
| // Concurrency, stated rather than solved: two converges in one process (the |
| // hourly RunAdopt and the version-change converge inside Pass) are |
| @@ -164,6 +188,20 @@ func Converge(o Ops, p Plan) (Result, error) { |
| } |
| |
| hostRole := p.Set.HostRole && p.Grants.HostRole |
| + res.Prev.Daemon = p.Set.Daemon |
| + |
| + // Not a refusal — a granted port on a hibernating route is a legitimate |
| + // shape (socket activation is the wake-on-connect half of hibernation) — |
| + // but it is a wake path that does NOT go through the activator, so the |
| + // edge's health gate does not cover it: a broken boot answers a connection |
| + // on :25 by parking it in the backlog rather than by returning a clean |
| + // 503. One line per pass, for as long as the combination exists, because |
| + // the operator who granted the port did not necessarily choose this. |
| + if p.Hibernate { |
| + for _, tok := range grantedPortTokens(p) { |
| + o.logf(" %s: %s is granted on a hibernating route — a connection to that port starts the unit directly, bypassing the activator's health-gated wake; the edge's clean 503 on a broken boot does not cover this path", p.Host, tok) |
| + } |
| + } |
| |
| // Read both halves BEFORE any refusal that needs to know what this route |
| // already claims, and before anything is written. Nothing here changes |
| @@ -213,40 +251,50 @@ func Converge(o Ops, p Plan) (Result, error) { |
| return res, err |
| } |
| |
| - svcChanged, err := o.materialize(p.Host, svcPath, svcBody, curSvc, svcExists, svcGenerated) |
| - if err != nil { |
| - return res, err |
| + svcChanged, svcErr := o.materialize(p.Host, svcPath, svcBody, curSvc, svcExists, svcGenerated) |
| + var sockChanged bool |
| + var sockErr error |
| + if svcErr == nil { |
| + sockChanged, sockErr = o.materialize(p.Host, sockPath, sockBody, curSock, sockExists, sockGenerated) |
| } |
| - sockChanged, err := o.materialize(p.Host, sockPath, sockBody, curSock, sockExists, sockGenerated) |
| - if err != nil { |
| - // The service half may already have been written; report it so the |
| - // caller's Changed is honest about what is on disk. |
| - res.Changed = svcChanged |
| - if svcChanged { |
| - _ = o.systemctl("daemon-reload") |
| - } |
| + res.Changed = svcChanged || sockChanged |
| + if err := errors.Join(svcErr, sockErr); err != nil { |
| + // A half-written pass is the divergence this whole rollback exists to |
| + // prevent: one unit converged, the other not, and the bytes on disk |
| + // claiming both are done. Put them back so the next pass sees a diff |
| + // and retries the WHOLE change. |
| + res.Changed = o.rollbackBytes(res.Prev, res.Changed) |
| return res, err |
| } |
| - res.Changed = svcChanged || sockChanged |
| |
| // systemd reads nothing from disk until it is told to. One reload, and |
| // only when something actually changed — this is what makes an unchanged |
| - // pass free. |
| + // pass free. It happens BEFORE any restart or start below, without |
| + // exception: a restart against configuration systemd has not re-read is a |
| + // pass that looks like it worked and changed nothing. |
| if res.Changed { |
| if err := o.systemctl("daemon-reload"); err != nil { |
| o.logf(" %s: daemon-reload: %v", p.Host, err) |
| + res.Changed = o.rollbackBytes(res.Prev, res.Changed) |
| return res, err |
| } |
| } |
| |
| sockLive := sockBody != nil |
| - restarted, err := o.applyUnitState(p, sockLive, svcChanged, sockChanged) |
| - if err != nil { |
| - return res, err |
| + st := o.applyUnitState(p, sockLive, svcChanged, sockChanged) |
| + if st.err != nil { |
| + if st.diverged { |
| + // The socket sequence is the one place where systemd's running |
| + // state can end up disagreeing with bytes that read "correct". |
| + // Roll them back so the next pass retries rather than concluding |
| + // there is nothing to do. |
| + res.Changed = o.rollbackBytes(res.Prev, res.Changed) |
| + } |
| + return res, st.err |
| } |
| if res.Changed { |
| what := "capability drop-ins converged" |
| - if restarted { |
| + if st.restarted { |
| what += " and the unit restarted" |
| } |
| o.logf(" %s: %s", p.Host, what) |
| @@ -254,15 +302,97 @@ func Converge(o Ops, p Plan) (Result, error) { |
| return res, nil |
| } |
| |
| +// rollbackBytes puts both halves back to what this pass found, WITHOUT |
| +// touching unit state — the caller is already in a failure path and the point |
| +// is only to stop the disk from claiming a change systemd never took. It |
| +// reports whether any byte is still moved afterwards, which is what Converge |
| +// returns as Changed: false when the disk is back where it started, true when |
| +// even the roll-back failed. |
| +func (o Ops) rollbackBytes(prev Snapshot, changed bool) bool { |
| + if !changed { |
| + return false |
| + } |
| + svcChanged, sockChanged, errs := o.applyHalves(prev) |
| + if len(errs) > 0 { |
| + o.logf(" %s: could not roll the capability drop-ins back after a failed pass (%v) — the next pass will find them mid-change", prev.Host, errors.Join(errs...)) |
| + return true |
| + } |
| + if svcChanged || sockChanged { |
| + if err := o.systemctl("daemon-reload"); err != nil { |
| + o.logf(" %s: daemon-reload after roll-back: %v", prev.Host, err) |
| + } |
| + o.logf(" %s: rolled the capability drop-ins back to what this pass found, so the next pass retries the whole change", prev.Host) |
| + } |
| + return false |
| +} |
| + |
| +// applyHalves writes a Snapshot's bytes over whatever is on disk now — the |
| +// shared body of Restore and rollbackBytes. Marker-checked in both |
| +// directions, exactly as Converge is: a half that was not ours when the |
| +// snapshot was taken is not ours to write over or delete now either. |
| +func (o Ops) applyHalves(want Snapshot) (svcChanged, sockChanged bool, errs []error) { |
| + svcPath, sockPath := DropInPaths(o.etcDir(), want.Host) |
| + halves := []struct { |
| + path string |
| + body []byte |
| + had bool |
| + out *bool |
| + }{ |
| + {svcPath, want.Service, want.ServiceHad, &svcChanged}, |
| + {sockPath, want.Socket, want.SocketHad, &sockChanged}, |
| + } |
| + for _, half := range halves { |
| + body := half.body |
| + if !half.had { |
| + body = nil |
| + } |
| + cur, exists, generated, err := readDropIn(half.path) |
| + if err != nil { |
| + o.logf(" %s: read %s: %v", want.Host, half.path, err) |
| + errs = append(errs, err) |
| + continue |
| + } |
| + did, err := o.materialize(want.Host, half.path, body, cur, exists, generated) |
| + if err != nil { |
| + errs = append(errs, err) |
| + } |
| + *half.out = did |
| + } |
| + return svcChanged, sockChanged, errs |
| +} |
| + |
| +// grantedPortTokens is the declared-AND-granted port set, as canonical grant |
| +// tokens, sorted — the ports that actually materialize. |
| +func grantedPortTokens(p Plan) []string { |
| + var toks []string |
| + for _, decl := range p.Set.Ports { |
| + if p.Grants.Ports[decl.Port] { |
| + toks = append(toks, PortToken(decl.Port)) |
| + } |
| + } |
| + sort.Strings(toks) |
| + return toks |
| +} |
| + |
| +// unitState is what applyUnitState did and how badly it failed. diverged is |
| +// the distinction that matters: a failed SOCKET SEQUENCE means systemd's |
| +// running listener no longer matches the bytes on disk, and only the caller |
| +// rolling those bytes back makes the next pass retry. |
| +type unitState struct { |
| + restarted bool |
| + diverged bool |
| + err error |
| +} |
| + |
| // applyUnitState is steps 7-9: the socket restart ordering, the enable/start |
| -// a daemon route's drift repair needs, and the service restart. It reports |
| -// whether the service was restarted. |
| -func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) (bool, error) { |
| +// a daemon route's drift repair needs, and the service restart. Every verb |
| +// here runs AFTER Converge's daemon-reload, never before it. |
| +func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) unitState { |
| svcUnit, sockUnit := ServiceUnit(p.Host), SocketUnit(p.Host) |
| - var firstErr error |
| + var st unitState |
| note := func(err error) { |
| - if err != nil && firstErr == nil { |
| - firstErr = err |
| + if err != nil && st.err == nil { |
| + st.err = err |
| } |
| } |
| |
| @@ -277,8 +407,11 @@ func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) (boo |
| // has to actually drop :25 — a rewrite nobody reloads into a running |
| // socket unit is a revoke that did nothing. |
| if sockChanged { |
| - err := o.socketSequence(p, "restart") |
| + err := o.socketSequence(p.Host, p.Set.Daemon, "restart") |
| note(err) |
| + if err != nil { |
| + st.diverged = true |
| + } |
| restartedService = true |
| } |
| |
| @@ -297,7 +430,7 @@ func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) (boo |
| // the service may be holding the port itself today (keymail binds |
| // :25 until its LISTEN_FDS change lands), and a bare socket start |
| // under it is the EADDRINUSE race again. |
| - note(o.socketSequence(p, "start")) |
| + note(o.socketSequence(p.Host, p.Set.Daemon, "start")) |
| restartedService = true |
| } |
| } |
| @@ -320,28 +453,43 @@ func (o Ops) applyUnitState(p Plan, sockLive, svcChanged, sockChanged bool) (boo |
| note(o.systemctl("restart", svcUnit)) |
| restartedService = true |
| } |
| - if firstErr != nil { |
| - o.logf(" %s: unit state: %v", p.Host, firstErr) |
| + st.restarted = restartedService |
| + if st.err != nil { |
| + o.logf(" %s: unit state: %v", p.Host, st.err) |
| } |
| - return restartedService, firstErr |
| + return st |
| } |
| |
| // socketSequence is the pinned ordering: stop the service, restart (or |
| -// start) the socket, start the service. The listener is down only for the |
| -// socket's own restart, during which inbound connections queue in the kernel |
| -// backlog — which is what makes spec §4c's "an MX restart never refuses a |
| -// connection" true rather than aspirational. |
| +// start) the socket, start the service. |
| +// |
| +// What the ordering buys, stated precisely so nobody reads more into it than |
| +// it gives: while the SERVICE is down and the socket unit is up, inbound |
| +// connections queue in the kernel backlog and are answered when the service |
| +// comes back — that is the phase spec §4c's "an MX restart never refuses a |
| +// connection" is about, and it is strictly better than today, where a relay |
| +// restart drops :25 entirely. During the socket unit's OWN restart the |
| +// listener does not exist, and connections in that window are refused, not |
| +// queued. The window is one systemctl verb wide and only opens when the |
| +// rendered ListenStream set actually changed. |
| // |
| // The service start is ATTEMPTED even when the socket verb failed. Returning |
| // early there would leave the service stopped — trading a socket that did not |
| // rebind for an app that is not running at all — so the error is remembered |
| // and returned, and the start happens anyway. |
| // |
| +// Accepted, and theoretical until keymail's LISTEN_FDS change lands: between |
| +// the stop and the socket verb the socket unit may still be up, so a |
| +// connection can socket-ACTIVATE the service back before this function starts |
| +// it — after which the start is systemd's no-op. That is a correct end state |
| +// (the service is running against the reloaded configuration), just not one |
| +// this function drove; it is not worth an isolate/mask dance to prevent. |
| +// |
| // A route that is neither a daemon nor currently running is neither stopped |
| // nor started: a hibernating tenant's drop-ins are unit text, not a wake. |
| -func (o Ops) socketSequence(p Plan, verb string) error { |
| - svcUnit, sockUnit := ServiceUnit(p.Host), SocketUnit(p.Host) |
| - shouldRun := p.Set.Daemon || o.query("is-active", svcUnit) |
| +func (o Ops) socketSequence(host string, daemon bool, verb string) error { |
| + svcUnit, sockUnit := ServiceUnit(host), SocketUnit(host) |
| + shouldRun := daemon || o.query("is-active", svcUnit) |
| |
| var errs []error |
| if shouldRun { |
| @@ -544,7 +692,7 @@ func scanClaims(etcDir, exceptHost string) (map[int]string, map[string]string) { |
| switch kind { |
| case "socket": |
| if v, ok := strings.CutPrefix(line, "ListenStream="); ok { |
| - if n, err := strconv.Atoi(v); err == nil { |
| + if n, ok := listenStreamPort(v); ok { |
| ports[n] = host |
| } |
| } |
| @@ -558,6 +706,30 @@ func scanClaims(etcDir, exceptHost string) (map[int]string, map[string]string) { |
| return ports, dirs |
| } |
| |
| +// listenStreamPort reads the port out of a ListenStream= value. This package |
| +// only ever WRITES the bare-number form, but the claim scan also reads files |
| +// this package did not write, and systemd accepts several spellings of the |
| +// same bind: "25", "0.0.0.0:25", "[::]:25", "1.2.3.4:25". Taking the segment |
| +// after the last colon covers all of them, so an address form cannot dodge |
| +// the box-wide port claim — the check would be worth little if it could. |
| +// |
| +// A value containing a path separator is a unix socket (the template's own |
| +// ListenStream=/run/carlos/apps/%i.sock) and claims no TCP port at all. |
| +func listenStreamPort(v string) (int, bool) { |
| + v = strings.TrimSpace(v) |
| + if v == "" || strings.Contains(v, "/") { |
| + return 0, false |
| + } |
| + if i := strings.LastIndex(v, ":"); i >= 0 { |
| + v = v[i+1:] |
| + } |
| + n, err := strconv.Atoi(v) |
| + if err != nil || n < 1 || n > 65535 { |
| + return 0, false |
| + } |
| + return n, true |
| +} |
| + |
| // parseUnitDir splits "carlos-app@<host>.service.d" into its host and kind. |
| func parseUnitDir(name string) (host, kind string, ok bool) { |
| rest, ok := strings.CutPrefix(name, "carlos-app@") |
| @@ -595,6 +767,18 @@ func dropInHas(body []byte, line string) bool { |
| // A plan whose Set is nil claims nothing: it is the leave-alone state, and |
| // leaving a drop-in alone is not a new claim. An ungranted port likewise |
| // claims nothing, because nothing materializes for it. |
| +// |
| +// Known limit, named rather than papered over: this tie-break is decided |
| +// against the PLANS, and Converge's own check is decided against what is |
| +// already on DISK. When the lexicographically-LATER host is the one already |
| +// materialized, the two disagree — CheckClaims refuses the later host, and |
| +// Converge then refuses the earlier host too, because the later host's |
| +// drop-in really is holding the port. The result is a stable both-refused |
| +// stall, not a flap: neither route materializes and both log every pass until |
| +// an operator revokes one grant. That is the fail-closed direction (nobody |
| +// silently takes :25 off a running mail server), and the alternative — |
| +// letting the winner evict the incumbent's drop-in — is a pass that stops |
| +// somebody else's listener to satisfy a sort order. |
| func CheckClaims(plans []Plan) map[string]error { |
| sorted := make([]Plan, len(plans)) |
| copy(sorted, plans) |
| @@ -634,12 +818,16 @@ func CheckClaims(plans []Plan) map[string]error { |
| return refused |
| } |
| |
| -// Restore puts prev back and daemon-reloads — the drop-in half of adoptHost's |
| -// restart-failure rollback (the symlink half is release.PointLive's). |
| +// Restore puts prev back — the drop-in half of adoptHost's restart-failure |
| +// rollback (the symlink half is release.PointLive's). |
| // |
| -// It restarts nothing: the caller that rolled the symlink back owns the |
| -// restart, and Restore's job is to make the unit text on disk match the |
| -// version the symlink now names again. |
| +// The SERVICE half's restart is the caller's: it rolled the symlink back and |
| +// it owns bringing the app up on the old version. The SOCKET half's is NOT. |
| +// Bytes are not a binding: a restored ListenStream that systemd never re-binds |
| +// restores nothing — the MX stays off :25 after a rollback — and a rolled-back |
| +// ADDED port stays publicly bound to a binary that never asked for it. So when |
| +// the socket half's bytes move, Restore runs the same stop → restart → start |
| +// sequence Converge does, which is why Snapshot carries Daemon. |
| // |
| // Marker-checked in both directions, exactly as Converge is: a half that was |
| // not ours when the snapshot was taken is not ours to write over or delete |
| @@ -649,41 +837,21 @@ func Restore(o Ops, prev Snapshot) error { |
| o.logf(" %q: %v", prev.Host, err) |
| return err |
| } |
| - svcPath, sockPath := DropInPaths(o.etcDir(), prev.Host) |
| - |
| - var errs []error |
| - changed := false |
| - for _, half := range []struct { |
| - path string |
| - body []byte |
| - had bool |
| - }{ |
| - {svcPath, prev.Service, prev.ServiceHad}, |
| - {sockPath, prev.Socket, prev.SocketHad}, |
| - } { |
| - want := half.body |
| - if !half.had { |
| - want = nil |
| - } |
| - cur, exists, generated, err := readDropIn(half.path) |
| - if err != nil { |
| - o.logf(" %s: read %s: %v", prev.Host, half.path, err) |
| - errs = append(errs, err) |
| - continue |
| - } |
| - did, err := o.materialize(prev.Host, half.path, want, cur, exists, generated) |
| - if err != nil { |
| - errs = append(errs, err) |
| - } |
| - changed = changed || did |
| - } |
| - if changed { |
| + svcChanged, sockChanged, errs := o.applyHalves(prev) |
| + if svcChanged || sockChanged { |
| + // Before any unit verb, without exception. |
| if err := o.systemctl("daemon-reload"); err != nil { |
| o.logf(" %s: daemon-reload: %v", prev.Host, err) |
| errs = append(errs, err) |
| } |
| o.logf(" %s: capability drop-ins rolled back to the previous version's", prev.Host) |
| } |
| + if sockChanged { |
| + if err := o.socketSequence(prev.Host, prev.Daemon, "restart"); err != nil { |
| + o.logf(" %s: rebinding the previous version's listeners: %v", prev.Host, err) |
| + errs = append(errs, err) |
| + } |
| + } |
| return errors.Join(errs...) |
| } |
| |
amadan