amadan

carlos / platform Public

zdd plan 1 task 4: the canonical socket path is a symlink to the current generation

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 43e325ad1842d80f27b3997952bb40abab07913b parent b98eed7
6 files changed, +427 −1
  • internal/activator/activator.go +16 −0
  • internal/activator/backend_exec.go +7 −0
  • internal/activator/canonlink.go +79 −0
  • internal/activator/canonlink_test.go +248 −0
  • internal/host/socket_advice.go +31 −1
  • internal/host/socket_advice_gen_test.go +46 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index d84b208b4..5758b5e6e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1022,6 +1022,14 @@ func (a *Service) wake(e *actEntry) {
if e.gen.N != 0 {
a.noteCurrent(e.host, e.gen.N)
}
+ // Outside consumers of the canonical path reach the generation
+ // that just became current (spec §7.1). A failed link costs them
+ // the canonical path, not the wake: internal traffic never uses it.
+ if a.gens && linksCanonical(e.route, e.gen) {
+ if err := swapCanonicalLink(e.route.Socket, e.gen.Socket); err != nil {
+ log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err)
+ }
+ }
if a.fences[e.host] != 0 {
a.beginStopLocked(e)
} else if e.bounceAfterWake {
@@ -2404,6 +2412,14 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
a.rememberStopLocked(e.host, outcome)
delete(a.entries, e.host)
retired := e.gen
+ // The canonical link goes with the last generation (spec §7.1). Under
+ // a.mu, the lock the awake transition creates links under, so a new
+ // entry's link cannot be removed by this one's stop.
+ if a.gens && linksCanonical(e.route, retired) {
+ if err := removeCanonicalLink(e.route.Socket, retired.Socket); err != nil {
+ log.Printf("activator: %s: remove link %s: %v", e.host, e.route.Socket, err)
+ }
+ }
a.mu.Unlock()
// The generation stopped being current when its entry went (spec §7.2:
// its idle connections close now, not at the transport's idle timeout).
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 1bedddea7..d37bfa371 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -1181,6 +1181,13 @@ func (b *ExecBackend) StartGen(ctx context.Context, rt registry.Route, n uint64,
case LayoutFenced, LayoutConflict:
return nil, Gen{}, fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
default:
+ if !perGen {
+ // A link left at the canonical path by an agent that ran with
+ // generations on would make the app's bind fail.
+ if err := clearLeftoverCanonicalLink(rt.Socket); err != nil {
+ return nil, Gen{}, fmt.Errorf("%s: leftover canonical link: %w", rt.Host, err)
+ }
+ }
if sock != rt.Socket {
if err := clearOwnGenSocket(sock); err != nil {
return nil, Gen{}, fmt.Errorf("%s: generation socket: %w", rt.Host, err)
diff --git a/internal/activator/canonlink.go b/internal/activator/canonlink.go
new file mode 100644
index 000000000..d545d567b
--- /dev/null
+++ b/internal/activator/canonlink.go
@@ -0,0 +1,79 @@
+package activator
+
+import (
+ "errors"
+ "os"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// canonlink.go keeps a route's canonical socket path (rt.Socket) pointing at
+// its current generation, for processes outside the edge and the agent that
+// dial it, such as members of the carlos-sockets group (#103; zero-downtime
+// spec §7.1, rev 15). Unix connect() follows the symlink. Internal dials
+// never use it: they go through AppSocket or a handle, and unixdial.Checked
+// would refuse the link anyway (the listener's bound path is the generation
+// path, not the dialled one).
+//
+// Plan 1 creates the link when the first generation becomes current and
+// removes it when the entry stops. Plan 3 calls swapCanonicalLink at every
+// switch.
+
+// linksCanonical reports whether g's process is reached through a symlink at
+// rt's canonical path: only when the generation bound a path of its own. A
+// generation on the canonical socket itself (generations off, a unit tenant,
+// a foreign-layout row) is never linked: the link would replace the real
+// socket.
+func linksCanonical(rt registry.Route, g Gen) bool {
+ return rt.Socket != "" && g.Socket != "" && g.Socket != rt.Socket
+}
+
+// swapCanonicalLink points canonical at target atomically: a temporary link
+// beside it, then rename over it. A dialler sees the old target or the new
+// one, never no path. Whatever was at canonical (an earlier link, or a stale
+// socket file from before generations) is replaced.
+func swapCanonicalLink(canonical, target string) error {
+ tmp := canonical + ".lnk"
+ if err := os.Remove(tmp); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ if err := os.Symlink(target, tmp); err != nil {
+ return err
+ }
+ if err := os.Rename(tmp, canonical); err != nil {
+ _ = os.Remove(tmp)
+ return err
+ }
+ return nil
+}
+
+// removeCanonicalLink removes canonical only while it is a symlink to
+// target: a link a later generation (or a later entry) has already moved,
+// and a real socket, are left alone.
+func removeCanonicalLink(canonical, target string) error {
+ got, err := os.Readlink(canonical)
+ if err != nil || got != target {
+ return nil
+ }
+ if err := os.Remove(canonical); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ return nil
+}
+
+// clearLeftoverCanonicalLink removes a symlink at canonical, whatever it
+// names, and nothing else. An agent with generations on that died without
+// stopping its entries leaves one; an app told to bind canonical with the
+// switch off would otherwise fail with "address in use". A canonical path
+// that is not a symlink (every box that never ran generations) is not
+// touched.
+func clearLeftoverCanonicalLink(canonical string) error {
+ fi, err := os.Lstat(canonical)
+ if err != nil || fi.Mode()&os.ModeSymlink == 0 {
+ return nil
+ }
+ if err := os.Remove(canonical); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ return nil
+}
diff --git a/internal/activator/canonlink_test.go b/internal/activator/canonlink_test.go
new file mode 100644
index 000000000..0149776c9
--- /dev/null
+++ b/internal/activator/canonlink_test.go
@@ -0,0 +1,248 @@
+package activator
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// serveOn binds an HTTP server at sock that answers body, the way a
+// generation's process would. It returns the server so a test can stop it.
+func serveOn(t *testing.T, sock, body string) *http.Server {
+ t.Helper()
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ if strings.Contains(err.Error(), "operation not permitted") {
+ t.Skipf("sandbox forbids unix sockets: %v", err)
+ }
+ t.Fatalf("listen %s: %v", sock, err)
+ }
+ srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { io.WriteString(w, body) })}
+ go srv.Serve(ln)
+ t.Cleanup(func() { srv.Close() })
+ return srv
+}
+
+// dialAsOutsider GETs / over a plain unix dial of path, the way a member of
+// the carlos-sockets group dials an instance: no unixdial checks, so the
+// kernel follows a symlink at path.
+func dialAsOutsider(t *testing.T, path string) string {
+ t.Helper()
+ c := &http.Client{Timeout: 2 * time.Second, Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ },
+ }}
+ resp, err := c.Get("http://carlos/")
+ if err != nil {
+ t.Fatalf("outside dial of %s: %v", path, err)
+ }
+ defer resp.Body.Close()
+ b, _ := io.ReadAll(resp.Body)
+ return string(b)
+}
+
+// servingProc is a fake child whose Stop also stops its server, so the
+// generation's socket goes away with it.
+type servingProc struct {
+ Proc
+ srv *http.Server
+}
+
+func (p *servingProc) Stop(ctx context.Context) error {
+ p.srv.Close()
+ return p.Proc.Stop(ctx)
+}
+
+// sockGenBackend is genFakeBackend whose generations really serve: perGen
+// binds <dir>/g<n>.sock, otherwise rt.Socket, each answering "gen <n>".
+type sockGenBackend struct {
+ *fakeBackend
+ t *testing.T
+ dir string
+}
+
+func (b *sockGenBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
+ proc, ver, err := b.fakeBackend.Start(ctx, rt)
+ if err != nil {
+ return proc, Gen{}, err
+ }
+ sock := rt.Socket
+ if perGen {
+ sock = filepath.Join(b.dir, fmt.Sprintf("g%d.sock", n))
+ }
+ srv := serveOn(b.t, sock, fmt.Sprintf("gen %d", n))
+ return &servingProc{Proc: proc, srv: srv}, Gen{N: n, Socket: sock, Control: sock + ".control", Build: ver}, nil
+}
+
+func (b *sockGenBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) {
+ return b.fakeBackend.NextDue(ctx, rt)
+}
+
+// sockGenActivator is a Service over sockGenBackend, its sockets in a short
+// directory (run with TMPDIR=/tmp/k), and a route whose canonical socket is
+// <dir>/c.sock.
+func sockGenActivator(t *testing.T, on bool) (*Service, registry.Route) {
+ t.Helper()
+ withGenerations(t, on)
+ dir := shortRunDir(t)
+ reg, err := registry.Open(filepath.Join(dir, "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, &sockGenBackend{fakeBackend: &fakeBackend{}, t: t, dir: dir}, reg)
+ rt := testRoute("link.example.com")
+ rt.Socket = filepath.Join(dir, "c.sock")
+ return a, rt
+}
+
+func isSymlink(path string) bool {
+ fi, err := os.Lstat(path)
+ return err == nil && fi.Mode()&os.ModeSymlink != 0
+}
+
+// The helper plan 3 calls at every switch: it replaces whatever is at the
+// canonical path, and leaves no temporary link behind.
+func TestSwapCanonicalLinkReplacesAtomically(t *testing.T) {
+ dir := t.TempDir()
+ canonical := filepath.Join(dir, "c.sock")
+ if err := os.WriteFile(canonical, []byte("stale socket file"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ for _, target := range []string{"/run/carlos/g/aaaaaaaaaaaa.1.sock", "/run/carlos/g/aaaaaaaaaaaa.2.sock"} {
+ if err := swapCanonicalLink(canonical, target); err != nil {
+ t.Fatalf("swapCanonicalLink(%s): %v", target, err)
+ }
+ if got, err := os.Readlink(canonical); err != nil || got != target {
+ t.Fatalf("canonical names %q (%v), want %q", got, err, target)
+ }
+ }
+ if _, err := os.Lstat(canonical + ".lnk"); !os.IsNotExist(err) {
+ t.Fatalf("temporary link left behind (lstat: %v)", err)
+ }
+}
+
+// A process dialling the canonical path follows the link to whichever
+// generation it names, before and after a switch.
+func TestSwapCanonicalLinkMovesOutsideDiallers(t *testing.T) {
+ dir := shortRunDir(t)
+ g1, g2 := filepath.Join(dir, "g1.sock"), filepath.Join(dir, "g2.sock")
+ serveOn(t, g1, "one")
+ serveOn(t, g2, "two")
+ canonical := filepath.Join(dir, "c.sock")
+ if err := swapCanonicalLink(canonical, g1); err != nil {
+ t.Fatal(err)
+ }
+ if got := dialAsOutsider(t, canonical); got != "one" {
+ t.Fatalf("before the switch: %q, want one", got)
+ }
+ if err := swapCanonicalLink(canonical, g2); err != nil {
+ t.Fatal(err)
+ }
+ if got := dialAsOutsider(t, canonical); got != "two" {
+ t.Fatalf("after the switch: %q, want two", got)
+ }
+}
+
+func TestRemoveCanonicalLinkRemovesOnlyItsOwnLink(t *testing.T) {
+ dir := t.TempDir()
+ canonical := filepath.Join(dir, "c.sock")
+ if err := swapCanonicalLink(canonical, "/a"); err != nil {
+ t.Fatal(err)
+ }
+ if err := removeCanonicalLink(canonical, "/b"); err != nil || !isSymlink(canonical) {
+ t.Fatalf("a link to another generation was removed (err %v)", err)
+ }
+ if err := removeCanonicalLink(canonical, "/a"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Lstat(canonical); !os.IsNotExist(err) {
+ t.Fatalf("own link not removed (lstat: %v)", err)
+ }
+ if err := os.WriteFile(canonical, nil, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := removeCanonicalLink(canonical, "/a"); err != nil || !isRegular(canonical) {
+ t.Fatalf("a real file at the canonical path was removed (err %v)", err)
+ }
+}
+
+// Spec §7.1 rev 15, end to end through the Service: the link appears when
+// the generation becomes current, an outside dialler reaches that
+// generation through it, and it goes when the entry stops.
+func TestCanonicalPathLinksToTheCurrentGeneration(t *testing.T) {
+ requireUnixSockets(t)
+ a, rt := sockGenActivator(t, true)
+ arrangeAwake(t, a, rt)
+ g, _ := a.ServingGen(rt.Host)
+ if target, err := os.Readlink(rt.Socket); err != nil || target != g.Socket {
+ t.Fatalf("canonical path names %q (%v), want the current generation %q", target, err, g.Socket)
+ }
+ if got, want := dialAsOutsider(t, rt.Socket), fmt.Sprintf("gen %d", g.N); got != want {
+ t.Fatalf("outside dial of the canonical path got %q, want %q", got, want)
+ }
+ stopAndWait(t, a, rt.Host)
+ if _, err := os.Lstat(rt.Socket); !os.IsNotExist(err) {
+ t.Fatalf("canonical link outlived its entry (lstat: %v)", err)
+ }
+}
+
+// Flag off: the canonical path is the app's own socket, never a link.
+func TestCanonicalPathIsARealSocketWithGenerationsOff(t *testing.T) {
+ requireUnixSockets(t)
+ a, rt := sockGenActivator(t, false)
+ arrangeAwake(t, a, rt)
+ if isSymlink(rt.Socket) || !isSocket(rt.Socket) {
+ t.Fatalf("%s is not a real socket with generations off", rt.Socket)
+ }
+ g, _ := a.ServingGen(rt.Host)
+ if got, want := dialAsOutsider(t, rt.Socket), fmt.Sprintf("gen %d", g.N); got != want {
+ t.Fatalf("outside dial got %q, want %q", got, want)
+ }
+}
+
+// A generation that IS on the canonical socket (a unit tenant, a backend
+// without generations) is never linked, even with the switch on: the link
+// would replace the real socket.
+func TestNoCanonicalLinkWhenTheGenerationOwnsTheCanonicalSocket(t *testing.T) {
+ withGenerations(t, true)
+ a, _, _, _ := testActivator(t, time.Hour)
+ rt := genTestRoute(t, "plain-link.example.com")
+ arrangeAwake(t, a, rt)
+ if _, err := os.Lstat(rt.Socket); !os.IsNotExist(err) {
+ t.Fatalf("something was created at the canonical path (lstat: %v)", err)
+ }
+}
+
+// Flag off, after an agent that ran with generations on died without
+// stopping its entries: the leftover link is cleared so the app can bind.
+func TestStartGenWithGenerationsOffClearsALeftoverCanonicalLink(t *testing.T) {
+ requireUnixSockets(t)
+ b := testExecBackend(t)
+ b.RunDirOverride = shortRunDir(t)
+ host := "leftover.example.com"
+ stageServingLive(t, b, host)
+ rt := flatGenRoute(b, host)
+ if err := os.Symlink(filepath.Join(b.RunDirOverride, "g", "gone.1.sock"), rt.Socket); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ proc, _, err := b.StartGen(ctx, rt, 1, false)
+ if err != nil {
+ t.Fatalf("StartGen over a leftover link: %v", err)
+ }
+ defer proc.Stop(context.Background())
+ if !isSocket(rt.Socket) {
+ t.Fatalf("%s is not the app's socket", rt.Socket)
+ }
+}
diff --git a/internal/host/socket_advice.go b/internal/host/socket_advice.go
index 7f17602f8..366be32e0 100644
--- a/internal/host/socket_advice.go
+++ b/internal/host/socket_advice.go
@@ -12,6 +12,7 @@ import (
"log"
"math/rand/v2"
"net"
+ "os"
"time"
"github.com/carlosframework/platform/internal/activator"
@@ -70,6 +71,31 @@ func socketServing(socket string) bool {
return true
}
+// advisorySocketServing is the advisory's probe with generations on. A
+// managed instance's canonical path is then the activator's own symlink to
+// its current generation (zero-downtime spec §7.1), which socketServing's
+// checked dial would report as a peer mismatch every hour. A link naming
+// the host's current generation is the activator's and is not a stray;
+// every other path is probed as before. appSocket is the activator's.
+func advisorySocketServing(routes []registry.Route, appSocket func(host string) (string, bool), serving func(string) bool) func(string) bool {
+ hostOf := map[string]string{}
+ for _, r := range routes {
+ if r.Kind == "instance" && r.Hibernate && r.Socket != "" && r.Backing != registry.BackingUnit {
+ hostOf[r.Socket] = r.Host
+ }
+ }
+ return func(sock string) bool {
+ if host, ok := hostOf[sock]; ok {
+ if cur, ok := appSocket(host); ok {
+ if target, err := os.Readlink(sock); err == nil && target == cur {
+ return false
+ }
+ }
+ }
+ return serving(sock)
+ }
+}
+
// socketAdvisoryLoop runs the advisory hourly, first pass ~1min after
// boot, like its sibling loops. svc may be nil (activator unarmed):
// owns is then constantly false and every live socket on an exec route
@@ -87,7 +113,11 @@ func socketAdvisoryLoop(reg *registry.Registry, svc *activator.Service) {
log.Printf("agent: socket advisory: list routes: %v", err)
continue
}
- for _, rt := range warnExternalSockets(routes, socketServing, owns) {
+ serving := socketServing
+ if svc != nil && svc.GenerationsOn() {
+ serving = advisorySocketServing(routes, svc.AppSocket, socketServing)
+ }
+ for _, rt := range warnExternalSockets(routes, serving, owns) {
if rt.Hibernate {
log.Printf("agent: route %s is activator-managed (hibernate=1) but its socket %s is served by a process the activator did not spawn — a stray supervisor is squatting the socket and will fight the wake path; find and retire it (see platform issue #17)", rt.Host, rt.Socket)
} else {
diff --git a/internal/host/socket_advice_gen_test.go b/internal/host/socket_advice_gen_test.go
new file mode 100644
index 000000000..162376c89
--- /dev/null
+++ b/internal/host/socket_advice_gen_test.go
@@ -0,0 +1,46 @@
+package host
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// With generations on, the advisory does not dial the activator's own link
+// at a managed route's canonical path (the checked dial would call it a peer
+// mismatch every hour). Any other link, or a path that is not a link, is
+// probed as before.
+func TestSocketAdvisorySkipsTheActivatorsOwnLink(t *testing.T) {
+ dir := t.TempDir()
+ own := filepath.Join(dir, "own.sock")
+ foreign := filepath.Join(dir, "foreign.sock")
+ plain := filepath.Join(dir, "plain.sock")
+ for path, target := range map[string]string{own: "/run/carlos/g/aaaaaaaaaaaa.3.sock", foreign: "/run/carlos/elsewhere.sock"} {
+ if err := os.Symlink(target, path); err != nil {
+ t.Fatal(err)
+ }
+ }
+ routes := []registry.Route{
+ {Host: "own.example.com", Kind: "instance", Hibernate: true, Socket: own},
+ {Host: "foreign.example.com", Kind: "instance", Hibernate: true, Socket: foreign},
+ {Host: "plain.example.com", Kind: "instance", Hibernate: true, Socket: plain},
+ }
+ appSocket := func(string) (string, bool) { return "/run/carlos/g/aaaaaaaaaaaa.3.sock", true }
+ var probed []string
+ serving := advisorySocketServing(routes, appSocket, func(sock string) bool {
+ probed = append(probed, sock)
+ return true
+ })
+ if serving(own) {
+ t.Fatal("the activator's own link was reported as served by a stray")
+ }
+ if !serving(foreign) || !serving(plain) {
+ t.Fatal("a foreign link or a plain path was not probed")
+ }
+ if fmt.Sprint(probed) != fmt.Sprint([]string{foreign, plain}) {
+ t.Fatalf("probed %v, want the foreign link and the plain path only", probed)
+ }
+}