amadan

carlos / platform Public

host, registry: root never opens or chowns the service user's registry files through a link (#267)

Root's registry open, migration and creation run in a hidden
__registry-child under the service user's uid/gid with no supplementary
groups. Remaining ownership repair is O_NOFOLLOW, checked plain single-link,
fchown on the descriptor. Bootstrap creates O_CREAT|O_EXCL|O_NOFOLLOW and
replaces chown -R with an openat2 walker that refuses symlinks, multi-link
and special files and does not cross mounts. Root-only credential and mount
tests skip without root; run them on a dev box before rollout.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude dfc5507f722c373cfb0f76824953809d665d7f1d parent 869d4bc
26 files changed, +2165 −126
  • internal/host/agent.go +2 −6
  • internal/host/bootstrap.go +53 −28
  • internal/host/bootstrapfs_linux.go +165 −0
  • internal/host/bootstrapfs_linux_test.go +180 −0
  • internal/host/bootstrapfs_other.go +20 −0
  • internal/host/registry_privilege_test.go +134 −0
  • internal/host/registrychild_other.go +16 −0
  • internal/host/registrychild_unix.go +84 −0
  • internal/host/registrychild_unix_test.go +158 −0
  • internal/host/registryowner.go +32 −67
  • internal/host/registryowner_chown_unix.go +39 −2
  • internal/host/registryowner_descriptor_linux.go +16 −0
  • internal/host/registryowner_descriptor_unix.go +16 −0
  • internal/host/registryowner_test.go +8 −16
  • internal/host/systemapply.go +1 −4
  • internal/registry/claimhold.go +5 −0
  • internal/registry/claimrows.go +18 −0
  • internal/registry/claimtimers.go +15 −0
  • internal/registry/coldhandover.go +5 −0
  • internal/registry/fleets.go +50 −0
  • internal/registry/registry.go +329 −3
  • internal/registry/remote.go +243 −0
  • internal/registry/remote_dispatch.go +318 −0
  • internal/registry/remote_test.go +177 −0
  • internal/registry/timers.go +75 −0
  • main.go +6 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index c3afd0949..8899f0441 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -3974,12 +3974,8 @@ func runAdoptWithEnsureSystemUnits(args []string, buildClaims ClaimFactory, ensu
if !*restartOnly {
ensureUnits()
}
- // openServiceRegistry, not registry.Open: this pass runs as root
- // (carlos-adopt/-stage/-restart/-config-sync have no User=), opening the
- // registry is what creates its -wal/-shm, and the helper hands them back
- // to the user the edge runs as as part of the open. That matters most for
- // -restart-only, which returns a dozen lines below — see registryowner.go
- // for why a converge on every pass is what it takes.
+ // Root keeps SQLite in the service-user child for every adoption mode,
+ // including the early return for restart-only.
if !claimsEnabled {
reg, err = openServiceRegistry(*regPath)
if err != nil {
diff --git a/internal/host/bootstrap.go b/internal/host/bootstrap.go
index 4693ea852..90082bcb0 100644
--- a/internal/host/bootstrap.go
+++ b/internal/host/bootstrap.go
@@ -23,6 +23,7 @@ import (
"os/exec"
"os/user"
"path/filepath"
+ "strconv"
"strings"
)
@@ -460,13 +461,8 @@ func bootstrapWith(root string, consumers []string, planJSON bool) error {
{filepath.Join(root, "etc", "sysusers.d"), 0o755},
}
for _, d := range dirs {
- if err := os.MkdirAll(d.path, d.mode); err != nil {
- return fmt.Errorf("mkdir %s: %v", d.path, err)
- }
- // MkdirAll's mode is masked by umask, and a pre-existing dir keeps its
- // old mode; force it so re-runs converge (Eleven's staging lesson).
- if err := os.Chmod(d.path, d.mode); err != nil {
- return fmt.Errorf("chmod %s: %v", d.path, err)
+ if err := bootstrapMakeDir(d.path, d.mode); err != nil {
+ return fmt.Errorf("mkdir/chmod %s: %v", d.path, err)
}
}
@@ -479,7 +475,7 @@ func bootstrapWith(root string, consumers []string, planJSON bool) error {
// file owns it, and an edge running as serviceUser cannot write a
// root-owned registry: that is discussion #8, three region boxes left
// unstartable on 2026-08-31. Creating it here puts it inside the
- // `chown -R /data` that privilegedPlan already runs, so it is
+ // descriptor-based ownership walk that privilegedPlan runs, so it is
// serviceUser's before anything can race for it.
//
// An empty file is a valid empty SQLite database — the first opener
@@ -498,14 +494,11 @@ func bootstrapWith(root string, consumers []string, planJSON bool) error {
// empty it. The script lives in the private platform-infrastructure repo;
// whoever widens what it copies has to exclude data/.
regPath := filepath.Join(dataDir, "registry.db")
- if _, err := os.Stat(regPath); os.IsNotExist(err) {
- f, err := os.OpenFile(regPath, os.O_CREATE|os.O_WRONLY, 0o640)
- if err != nil {
- return fmt.Errorf("create %s: %v", regPath, err)
- }
- if err := f.Close(); err != nil {
- return fmt.Errorf("create %s: %v", regPath, err)
- }
+ created, err := bootstrapCreateRegistry(regPath)
+ if err != nil {
+ return fmt.Errorf("create %s: %v", regPath, err)
+ }
+ if created {
fmt.Fprintf(out, " created %s\n", regPath)
}
@@ -633,7 +626,7 @@ func unitNames() ([]string, error) {
}
// privilegedPlan is the exact non-systemctl privileged sequence a live
-// bootstrap runs — and the same sequence a staging run PRINTS instead of
+// bootstrap dispatches — and the same sequence a staging run PRINTS instead of
// running (systemctlPlan's discipline, extended to the user/group half when
// issue #103 added the consumers group): the two can never drift into
// describing different behaviour.
@@ -641,11 +634,9 @@ func unitNames() ([]string, error) {
// The sequence: service user, consumers group, ownership. Every chown
// converges TOWARD the target state, never through an intermediate that
// undoes a previous run's work: /run/carlos is recursively
-// serviceUser:socketsGroup — recursive on purpose, so a re-run (the
-// documented way to enroll another consumer) also converges the group on
-// every LIVE instance socket instead of resetting it to serviceUser's group
-// and stranding the already-enrolled consumers until each instance rebinds —
-// and apps/ (unit-tenant sockets, systemd-bound, no path consumers) is then
+// serviceUser:socketsGroup. Unsafe entries, including live sockets, are
+// logged and skipped; their creating units establish socket ownership.
+// The apps/ directory (unit-tenant sockets, no path consumers) is then
// put back to serviceUser:serviceUser. The chmod re-asserts the setgid bit
// the tmpfiles line promises; consumers enroll last, one usermod each
// (`-aG` is append — an existing member is a no-op, so re-runs converge).
@@ -671,8 +662,23 @@ func privilegedPlan(consumers []string) [][]string {
// guards that keep the raw commands idempotent: useradd and groupadd exit
// non-zero when their subject already exists, and "already provisioned" is
// bootstrap's normal re-run state, not an error. Everything else in the plan
-// is idempotent as-is.
+// is idempotent. Ownership and mode steps use descriptor helpers; the
+// command text is retained for staging output and is never sent to chown.
func runPrivileged(args []string) error {
+ if len(args) == 4 && args[0] == "chown" && args[1] == "-R" {
+ owner := strings.SplitN(args[2], ":", 2)
+ if len(owner) != 2 {
+ return fmt.Errorf("invalid ownership %q", args[2])
+ }
+ uid, gid, err := bootstrapOwner(owner[0], owner[1])
+ if err != nil {
+ return err
+ }
+ return bootstrapChownTree(args[3], uid, gid)
+ }
+ if len(args) == 3 && args[0] == "chmod" && args[1] == "2750" {
+ return bootstrapMakeDir(args[2], 0o750|os.ModeSetgid)
+ }
// The subject is derived from the command line itself (both commands
// take it as their final argument), never from a package constant — a
// guard keyed to a constant would silently skip any SECOND useradd or
@@ -694,14 +700,33 @@ func runPrivileged(args []string) error {
return nil
}
-// chownToService recursively gives a path to the service user (offcloud's
-// credentials file; the standard dirs go through privilegedPlan).
+func bootstrapOwner(owner, group string) (int, int, error) {
+ u, err := user.Lookup(owner)
+ if err != nil {
+ return 0, 0, err
+ }
+ g, err := user.LookupGroup(group)
+ if err != nil {
+ return 0, 0, err
+ }
+ uid, err := strconv.Atoi(u.Uid)
+ if err != nil {
+ return 0, 0, err
+ }
+ gid, err := strconv.Atoi(g.Gid)
+ return uid, gid, err
+}
+
func chownToService(path string) error {
- out, err := exec.Command("chown", "-R", serviceUser+":"+serviceUser, path).CombinedOutput()
+ uid, gid, err := bootstrapOwner(serviceUser, serviceUser)
+ if err != nil {
+ return err
+ }
+ _, err = ownRegistryFile(path, uid, gid)
if err != nil {
- return fmt.Errorf("chown %s: %v: %s", path, err, strings.TrimSpace(string(out)))
+ log.Printf("ALERT: bootstrap: refusing ownership of %s: %v", path, err)
}
- return nil
+ return err
}
func systemctl(args ...string) error {
diff --git a/internal/host/bootstrapfs_linux.go b/internal/host/bootstrapfs_linux.go
new file mode 100644
index 000000000..536952eac
--- /dev/null
+++ b/internal/host/bootstrapfs_linux.go
@@ -0,0 +1,165 @@
+//go:build linux
+
+package host
+
+import (
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "golang.org/x/sys/unix"
+)
+
+func bootstrapOpenDir(path string) (int, error) {
+ return unix.Openat2(unix.AT_FDCWD, path, &unix.OpenHow{
+ Flags: unix.O_RDONLY | unix.O_DIRECTORY | unix.O_CLOEXEC | unix.O_NOFOLLOW,
+ Resolve: unix.RESOLVE_NO_SYMLINKS,
+ })
+}
+
+// bootstrapMakeDir anchors every component before creating or changing it.
+func bootstrapMakeDir(path string, mode os.FileMode) error {
+ absolute, err := filepath.Abs(path)
+ if err != nil {
+ return err
+ }
+ fd, err := bootstrapOpenDir("/")
+ if err != nil {
+ return err
+ }
+ defer func() { unix.Close(fd) }()
+ parts := strings.Split(strings.TrimPrefix(absolute, "/"), "/")
+ for _, name := range parts {
+ if name == "" {
+ continue
+ }
+ next, err := unix.Openat2(fd, name, &unix.OpenHow{Flags: unix.O_RDONLY | unix.O_DIRECTORY | unix.O_CLOEXEC | unix.O_NOFOLLOW, Resolve: unix.RESOLVE_BENEATH | unix.RESOLVE_NO_SYMLINKS})
+ if errors.Is(err, unix.ENOENT) {
+ if err = unix.Mkdirat(fd, name, uint32(mode.Perm())); err != nil && !errors.Is(err, unix.EEXIST) {
+ return err
+ }
+ next, err = unix.Openat2(fd, name, &unix.OpenHow{Flags: unix.O_RDONLY | unix.O_DIRECTORY | unix.O_CLOEXEC | unix.O_NOFOLLOW, Resolve: unix.RESOLVE_BENEATH | unix.RESOLVE_NO_SYMLINKS})
+ }
+ if err != nil {
+ return err
+ }
+ unix.Close(fd)
+ fd = next
+ }
+ bits := uint32(mode.Perm())
+ if mode&os.ModeSetgid != 0 {
+ bits |= unix.S_ISGID
+ }
+ if mode&os.ModeSetuid != 0 {
+ bits |= unix.S_ISUID
+ }
+ if mode&os.ModeSticky != 0 {
+ bits |= unix.S_ISVTX
+ }
+ return unix.Fchmod(fd, bits)
+}
+
+func bootstrapCreateRegistry(path string) (bool, error) {
+ parent, err := bootstrapOpenDir(filepath.Dir(path))
+ if err != nil {
+ return false, err
+ }
+ defer unix.Close(parent)
+ fd, err := unix.Openat2(parent, filepath.Base(path), &unix.OpenHow{
+ Flags: unix.O_CREAT | unix.O_EXCL | unix.O_WRONLY | unix.O_NOFOLLOW | unix.O_CLOEXEC,
+ Mode: 0o640,
+ Resolve: unix.RESOLVE_BENEATH | unix.RESOLVE_NO_SYMLINKS | unix.RESOLVE_NO_XDEV,
+ })
+ if errors.Is(err, unix.EEXIST) {
+ var st unix.Stat_t
+ if err := unix.Fstatat(parent, filepath.Base(path), &st, unix.AT_SYMLINK_NOFOLLOW); err != nil {
+ return false, err
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 {
+ log.Printf("ALERT: bootstrap: refusing unsafe registry file %s", path)
+ }
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ defer unix.Close(fd)
+ var st unix.Stat_t
+ if err := unix.Fstat(fd, &st); err != nil {
+ return false, err
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 {
+ return false, fmt.Errorf("unsafe registry file %s", path)
+ }
+ return true, nil
+}
+
+func bootstrapChownTree(path string, uid, gid int) error {
+ fd, err := bootstrapOpenDir(path)
+ if err != nil {
+ return err
+ }
+ return bootstrapChownDir(os.NewFile(uintptr(fd), path), uid, gid)
+}
+
+func bootstrapChownDir(dir *os.File, uid, gid int) error {
+ defer dir.Close()
+ // The name '.' is resolved against the opened directory, not its mutable parent.
+ if err := unix.Fchownat(int(dir.Fd()), ".", uid, gid, unix.AT_SYMLINK_NOFOLLOW); err != nil {
+ return err
+ }
+ entries, err := dir.ReadDir(-1)
+ if err != nil {
+ return err
+ }
+ for _, entry := range entries {
+ path := filepath.Join(dir.Name(), entry.Name())
+ // NO_XDEV refuses bind mounts as well as different-device mounts.
+ fd, err := unix.Openat2(int(dir.Fd()), entry.Name(), &unix.OpenHow{
+ Flags: unix.O_PATH | unix.O_NOFOLLOW | unix.O_CLOEXEC,
+ Resolve: unix.RESOLVE_BENEATH | unix.RESOLVE_NO_SYMLINKS | unix.RESOLVE_NO_XDEV,
+ })
+ if err != nil {
+ if errors.Is(err, unix.ELOOP) || errors.Is(err, unix.EXDEV) || errors.Is(err, unix.ENOENT) || errors.Is(err, unix.ENXIO) {
+ log.Printf("ALERT: bootstrap: skipping unsafe ownership path %s: %v", path, err)
+ continue
+ }
+ return err
+ }
+ f := os.NewFile(uintptr(fd), path)
+ var st unix.Stat_t
+ if err := unix.Fstat(fd, &st); err != nil {
+ f.Close()
+ return err
+ }
+ switch st.Mode & unix.S_IFMT {
+ case unix.S_IFDIR:
+ child, err := unix.Openat2(fd, ".", &unix.OpenHow{Flags: unix.O_RDONLY | unix.O_DIRECTORY | unix.O_CLOEXEC | unix.O_NOFOLLOW, Resolve: unix.RESOLVE_NO_SYMLINKS | unix.RESOLVE_NO_XDEV})
+ f.Close()
+ if err != nil {
+ return err
+ }
+ if err := bootstrapChownDir(os.NewFile(uintptr(child), path), uid, gid); err != nil {
+ return err
+ }
+ case unix.S_IFREG:
+ if st.Nlink != 1 {
+ log.Printf("ALERT: bootstrap: skipping unsafe ownership path %s: multiple links", path)
+ f.Close()
+ continue
+ }
+ err := unix.Fchownat(fd, "", uid, gid, unix.AT_EMPTY_PATH|unix.AT_SYMLINK_NOFOLLOW)
+ f.Close()
+ if err != nil {
+ return err
+ }
+ default:
+ log.Printf("ALERT: bootstrap: skipping unsafe ownership path %s: special file", path)
+ f.Close()
+ }
+ }
+ return nil
+}
diff --git a/internal/host/bootstrapfs_linux_test.go b/internal/host/bootstrapfs_linux_test.go
new file mode 100644
index 000000000..54bcb6a87
--- /dev/null
+++ b/internal/host/bootstrapfs_linux_test.go
@@ -0,0 +1,180 @@
+//go:build linux
+
+package host
+
+import (
+ "bytes"
+ "errors"
+ "log"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "golang.org/x/sys/unix"
+)
+
+func TestBootstrapRegistryPlantedLinks(t *testing.T) {
+ for _, dangling := range []bool{false, true} {
+ t.Run(map[bool]string{false: "existing", true: "dangling"}[dangling], func(t *testing.T) {
+ root := t.TempDir()
+ victim := filepath.Join(t.TempDir(), "victim")
+ if !dangling {
+ if err := os.WriteFile(victim, []byte("preserve"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := os.Mkdir(filepath.Join(root, "data"), 0o750); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(victim, filepath.Join(root, "data", "registry.db")); err != nil {
+ t.Fatal(err)
+ }
+ if err := bootstrap(root, nil); err != nil {
+ t.Fatal(err)
+ }
+ got, err := os.ReadFile(victim)
+ if dangling {
+ if !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("dangling target created: %v", err)
+ }
+ } else if err != nil || string(got) != "preserve" {
+ t.Fatalf("victim changed: %q %v", got, err)
+ }
+ })
+ }
+}
+
+func TestBootstrapRefusesWritableParentSymlink(t *testing.T) {
+ for _, parent := range []string{"data", "data/.instances", "run/carlos"} {
+ t.Run(strings.ReplaceAll(parent, "/", "_"), func(t *testing.T) {
+ root, victim := t.TempDir(), t.TempDir()
+ if err := os.Chmod(victim, 0o711); err != nil {
+ t.Fatal(err)
+ }
+ path := filepath.Join(root, parent)
+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(victim, path); err != nil {
+ t.Fatal(err)
+ }
+ if err := bootstrap(root, nil); err == nil {
+ t.Fatal("accepted parent symlink")
+ }
+ st, err := os.Stat(victim)
+ if err != nil || st.Mode().Perm() != 0o711 {
+ t.Fatalf("victim mode changed: %v %v", st, err)
+ }
+ entries, err := os.ReadDir(victim)
+ if err != nil || len(entries) != 0 {
+ t.Fatalf("victim populated: %v %v", entries, err)
+ }
+ })
+ }
+}
+
+func TestBootstrapOwnershipWalkSkipsUnsafeEntries(t *testing.T) {
+ dir := t.TempDir()
+ victim := filepath.Join(t.TempDir(), "victim")
+ if err := os.WriteFile(victim, []byte("preserve"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(victim, 0o750|os.ModeSetuid|os.ModeSetgid); err != nil {
+ t.Fatal(err)
+ }
+ before, err := os.Stat(victim)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, name := range []string{"registry.db-wal", "registry.db-shm"} {
+ if err := os.Link(victim, filepath.Join(dir, name)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := os.Symlink(victim, filepath.Join(dir, "registry.db")); err != nil {
+ t.Fatal(err)
+ }
+ if err := unix.Mkfifo(filepath.Join(dir, "fifo"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ socket, err := net.Listen("unix", filepath.Join(dir, "socket"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer socket.Close()
+ if err := os.Mkdir(filepath.Join(dir, "nested"), 0o750); err != nil {
+ t.Fatal(err)
+ }
+ plain := filepath.Join(dir, "nested", "plain")
+ if err := os.WriteFile(plain, []byte("ordinary"), 0o640); err != nil {
+ t.Fatal(err)
+ }
+ var logs bytes.Buffer
+ old := log.Writer()
+ log.SetOutput(&logs)
+ defer log.SetOutput(old)
+ if err := bootstrapChownTree(dir, os.Getuid(), os.Getgid()); err != nil {
+ t.Fatal(err)
+ }
+ after, err := os.Stat(victim)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after.Mode() != before.Mode() {
+ t.Fatalf("hard-linked victim ownership operation cleared mode: %v -> %v", before.Mode(), after.Mode())
+ }
+ got, err := os.ReadFile(victim)
+ if err != nil || string(got) != "preserve" {
+ t.Fatalf("victim changed: %q %v", got, err)
+ }
+ for _, name := range []string{"registry.db", "registry.db-wal", "registry.db-shm", "fifo", "socket"} {
+ if !strings.Contains(logs.String(), filepath.Join(dir, name)) {
+ t.Errorf("missing alert for %s: %s", name, logs.String())
+ }
+ }
+ got, err = os.ReadFile(plain)
+ if err != nil || string(got) != "ordinary" {
+ t.Fatalf("plain file changed: %q %v", got, err)
+ }
+}
+
+func TestBootstrapOwnershipWalkDoesNotCrossBindMount(t *testing.T) {
+ if os.Geteuid() != 0 {
+ t.Skip("bind mount test requires root with CAP_SYS_ADMIN")
+ }
+ dir, source := t.TempDir(), t.TempDir()
+ target := filepath.Join(dir, "bind")
+ if err := os.Mkdir(target, 0o750); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(source, 0o750|os.ModeSetgid); err != nil {
+ t.Fatal(err)
+ }
+ if err := unix.Mount(source, target, "", unix.MS_BIND, ""); err != nil {
+ if errors.Is(err, unix.EPERM) || errors.Is(err, unix.EACCES) {
+ t.Skipf("bind mount test requires CAP_SYS_ADMIN: %v", err)
+ }
+ t.Fatal(err)
+ }
+ defer unix.Unmount(target, 0)
+ before, err := os.Stat(source)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := bootstrapChownTree(dir, 65534, 65534); err != nil {
+ t.Fatal(err)
+ }
+ after, err := os.Stat(source)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var sourceStat unix.Stat_t
+ if err := unix.Stat(source, &sourceStat); err != nil {
+ t.Fatal(err)
+ }
+ if before.Mode() != after.Mode() || sourceStat.Uid != uint32(os.Getuid()) {
+ t.Fatalf("crossed bind mount: mode %v -> %v, uid %d", before.Mode(), after.Mode(), sourceStat.Uid)
+ }
+}
diff --git a/internal/host/bootstrapfs_other.go b/internal/host/bootstrapfs_other.go
new file mode 100644
index 000000000..816cb91f8
--- /dev/null
+++ b/internal/host/bootstrapfs_other.go
@@ -0,0 +1,20 @@
+//go:build !linux
+
+package host
+
+import (
+ "fmt"
+ "os"
+)
+
+func bootstrapMakeDir(path string, mode os.FileMode) error {
+ return fmt.Errorf("bootstrap filesystem operations require Linux")
+}
+
+func bootstrapCreateRegistry(path string) (bool, error) {
+ return false, fmt.Errorf("bootstrap filesystem operations require Linux")
+}
+
+func bootstrapChownTree(path string, uid, gid int) error {
+ return fmt.Errorf("bootstrap filesystem operations require Linux")
+}
diff --git a/internal/host/registry_privilege_test.go b/internal/host/registry_privilege_test.go
new file mode 100644
index 000000000..a32453747
--- /dev/null
+++ b/internal/host/registry_privilege_test.go
@@ -0,0 +1,134 @@
+//go:build unix
+
+package host
+
+import (
+ "bytes"
+ "errors"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "os"
+ "os/user"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func TestRootRegistryRefusesPlantedFilesBeforeSQLite(t *testing.T) {
+ u, err := user.Current()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if u.Uid == "0" {
+ t.Skip("this test simulates root repair with the current non-root service identity; real root credentials have separate coverage")
+ }
+ for _, name := range []string{"symlink", "dangling-symlink", "hardlink-wal", "hardlink-shm", "symlink-parent"} {
+ t.Run(name, func(t *testing.T) {
+ dir := t.TempDir()
+ victim := filepath.Join(dir, "victim")
+ data := filepath.Join(dir, "data")
+ if err := os.Mkdir(data, 0700); err != nil {
+ t.Fatal(err)
+ }
+ path := filepath.Join(data, "registry.db")
+ if name != "dangling-symlink" {
+ if err := os.WriteFile(victim, nil, 0640); err != nil {
+ t.Fatal(err)
+ }
+ }
+ switch name {
+ case "symlink", "dangling-symlink":
+ err = os.Symlink(victim, path)
+ case "hardlink-wal":
+ err = os.Link(victim, path+"-wal")
+ case "hardlink-shm":
+ err = os.Link(victim, path+"-shm")
+ case "symlink-parent":
+ alias := filepath.Join(dir, "alias")
+ err = os.Symlink(data, alias)
+ path = filepath.Join(alias, "registry.db")
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ before, beforeErr := os.Stat(victim)
+ logged := captureLog(t)
+ reg, openErr := openRegistryConverging(0, u.Username, path)
+ if reg != nil {
+ reg.Close()
+ t.Fatal("unsafe registry reached SQLite")
+ }
+ if openErr == nil {
+ t.Fatal("unsafe registry accepted")
+ }
+ if !strings.Contains(strings.ToLower(logged()), "alert") {
+ t.Errorf("refused file had no alert: %s", logged())
+ }
+ after, afterErr := os.Stat(victim)
+ if errors.Is(beforeErr, os.ErrNotExist) {
+ if !errors.Is(afterErr, os.ErrNotExist) {
+ t.Fatalf("dangling link target created: %v", afterErr)
+ }
+ } else {
+ if afterErr != nil {
+ t.Fatal(afterErr)
+ }
+ beforeUID, beforeGID, _ := fileOwner(before)
+ afterUID, afterGID, _ := fileOwner(after)
+ if beforeUID != afterUID || beforeGID != afterGID || before.Mode() != after.Mode() || before.ModTime() != after.ModTime() {
+ t.Fatal("victim ownership, mode or modification time changed")
+ }
+ contents, err := os.ReadFile(victim)
+ if err != nil || !bytes.Equal(contents, []byte{}) {
+ t.Fatalf("victim changed: %q, %v", contents, err)
+ }
+ }
+ for _, suffix := range []string{"-wal", "-shm", "-journal"} {
+ if _, err := os.Lstat(victim + suffix); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("SQLite created victim sidecar %s: %v", suffix, err)
+ }
+ }
+ })
+ }
+}
+
+func TestPrivilegedRegistryEntryPointsUseServiceBoundary(t *testing.T) {
+ for _, tc := range []struct{ file, function string }{
+ {"agent.go", "runAdoptWithEnsureSystemUnits"},
+ {"systemapply.go", "openSystemRegistry"},
+ {"claimstop_run.go", "RunInstanceControl"},
+ } {
+ t.Run(tc.function, func(t *testing.T) {
+ f, err := parser.ParseFile(token.NewFileSet(), tc.file, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ found := false
+ for _, decl := range f.Decls {
+ fn, ok := decl.(*ast.FuncDecl)
+ if !ok || fn.Name.Name != tc.function {
+ continue
+ }
+ ast.Inspect(fn.Body, func(n ast.Node) bool {
+ call, ok := n.(*ast.CallExpr)
+ if !ok {
+ return true
+ }
+ if id, ok := call.Fun.(*ast.Ident); ok && id.Name == "openServiceRegistry" {
+ found = true
+ }
+ if sel, ok := call.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Open" {
+ if id, ok := sel.X.(*ast.Ident); ok && id.Name == "registry" {
+ t.Error("privileged caller directly opens SQLite")
+ }
+ }
+ return true
+ })
+ }
+ if !found {
+ t.Fatal("privileged caller does not cross the service registry boundary")
+ }
+ })
+ }
+}
diff --git a/internal/host/registrychild_other.go b/internal/host/registrychild_other.go
new file mode 100644
index 000000000..7b9215f72
--- /dev/null
+++ b/internal/host/registrychild_other.go
@@ -0,0 +1,16 @@
+//go:build !unix
+
+package host
+
+import (
+ "errors"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func openRegistryChild(string, int, int) (*registry.Registry, error) {
+ return nil, errors.New("registry: privilege-dropped workers require unix")
+}
+
+func RunRegistryChild([]string) error {
+ return errors.New("registry: privilege-dropped workers require unix")
+}
diff --git a/internal/host/registrychild_unix.go b/internal/host/registrychild_unix.go
new file mode 100644
index 000000000..fd5fde016
--- /dev/null
+++ b/internal/host/registrychild_unix.go
@@ -0,0 +1,84 @@
+//go:build unix
+
+package host
+
+import (
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strconv"
+ "syscall"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const registryChildCommand = "__registry-child"
+
+func openRegistryChild(path string, uid, gid int) (*registry.Registry, error) {
+ if uid <= 0 || gid < 0 {
+ return nil, errors.New("registry: invalid service credentials")
+ }
+ path, err := filepath.Abs(path)
+ if err != nil {
+ return nil, err
+ }
+ executable, err := os.Executable()
+ if err != nil {
+ return nil, err
+ }
+ cmd := exec.Command(executable, registryChildCommand, path, strconv.Itoa(uid), strconv.Itoa(gid))
+ return startRegistryChild(cmd, path, uid, gid)
+}
+
+func startRegistryChild(cmd *exec.Cmd, path string, uid, gid int) (*registry.Registry, error) {
+ if uid <= 0 || gid < 0 {
+ return nil, errors.New("registry: invalid service credentials")
+ }
+ cmd.Env = []string{}
+ if value, ok := os.LookupEnv("CARLOS_HOST_CLAIM"); ok {
+ cmd.Env = append(cmd.Env, "CARLOS_HOST_CLAIM="+value)
+ }
+ cmd.Dir = "/"
+ cmd.SysProcAttr = &syscall.SysProcAttr{Credential: &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid), Groups: []uint32{}}}
+ cmd.Stderr = os.Stderr
+ input, err := cmd.StdinPipe()
+ if err != nil {
+ return nil, err
+ }
+ output, err := cmd.StdoutPipe()
+ if err != nil {
+ input.Close()
+ return nil, err
+ }
+ if err = cmd.Start(); err != nil {
+ input.Close()
+ output.Close()
+ return nil, err
+ }
+ return registry.OpenRemote(path, input, output, cmd.Wait)
+}
+
+// RunRegistryChild is the private re-exec entry point, before CLI dispatch.
+func RunRegistryChild(args []string) error {
+ if len(args) != 3 {
+ return errors.New("registry: invalid child arguments")
+ }
+ uid, err := strconv.Atoi(args[1])
+ if err != nil {
+ return err
+ }
+ gid, err := strconv.Atoi(args[2])
+ if err != nil {
+ return err
+ }
+ groups, err := os.Getgroups()
+ if err != nil {
+ return err
+ }
+ if uid <= 0 || os.Getuid() != uid || os.Geteuid() != uid || os.Getgid() != gid || os.Getegid() != gid || len(groups) != 0 {
+ return fmt.Errorf("registry: child credentials were not dropped")
+ }
+ return registry.ServeRemote(args[0], os.Stdin, os.Stdout)
+}
diff --git a/internal/host/registrychild_unix_test.go b/internal/host/registrychild_unix_test.go
new file mode 100644
index 000000000..d6ba47cd1
--- /dev/null
+++ b/internal/host/registrychild_unix_test.go
@@ -0,0 +1,158 @@
+//go:build unix
+
+package host
+
+import (
+ "errors"
+ "io"
+ "os"
+ "os/exec"
+ "os/user"
+ "path/filepath"
+ "strconv"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestRegistryChildHelper(t *testing.T) {
+ for i, arg := range os.Args {
+ if arg == "--registry-test-child" {
+ if err := RunRegistryChild(os.Args[i+1:]); err != nil {
+ os.Exit(1)
+ }
+ os.Exit(0)
+ }
+ }
+}
+
+func TestRegistryChildDropsCredentials(t *testing.T) {
+ if os.Geteuid() != 0 {
+ t.Skip("requires root to set the child uid/gid and clear supplementary groups")
+ }
+ t.Setenv("CARLOS_HOST_CLAIM", "true")
+ service, err := user.Lookup("nobody")
+ if err != nil {
+ t.Skip("requires an unprivileged nobody account")
+ }
+ uid, err := strconv.Atoi(service.Uid)
+ if err != nil || uid <= 0 {
+ t.Skip("requires a nonzero service uid")
+ }
+ gid, err := strconv.Atoi(service.Gid)
+ if err != nil {
+ t.Fatal(err)
+ }
+ dir, err := os.MkdirTemp(os.TempDir(), "registry-child-")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ if err := os.Chmod(dir, 0755); err != nil {
+ t.Fatal(err)
+ }
+ executable, err := os.Executable()
+ if err != nil {
+ t.Fatal(err)
+ }
+ source, err := os.Open(executable)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer source.Close()
+ bin := filepath.Join(dir, "test-worker")
+ dest, err := os.OpenFile(bin, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0755)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = io.Copy(dest, source)
+ closeErr := dest.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if closeErr != nil {
+ t.Fatal(closeErr)
+ }
+ data := filepath.Join(dir, "data")
+ if err := os.Mkdir(data, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chown(data, uid, gid); err != nil {
+ t.Fatal(err)
+ }
+ path := filepath.Join(data, "registry.db")
+ legacy, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := legacy.Close(); err != nil {
+ t.Fatal(err)
+ }
+ changed, err := ownRegistryFile(path, uid, gid)
+ if err != nil || !changed {
+ t.Fatalf("repair root-owned registry: changed=%v, err=%v", changed, err)
+ }
+ cmd := exec.Command(bin, "-test.run=^TestRegistryChildHelper$", "--", "--registry-test-child", path, service.Uid, service.Gid)
+ r, err := startRegistryChild(cmd, path, uid, gid)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ if err := r.Upsert(registry.Route{Host: "dropped", Kind: "instance"}); err != nil {
+ r.Close()
+ t.Fatal(err)
+ }
+ owner := registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"}
+ row, err := r.Get("dropped")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := r.AdoptClaimRoute(t.Context(), *row, owner); err != nil {
+ t.Fatal(err)
+ }
+ timer := registry.Timer{Host: "dropped", Name: "tick", Kind: "cron", Spec: "* * * * *"}
+ if err := r.UpsertTimer(timer); err != nil {
+ t.Fatal(err)
+ }
+ saved, err := r.GetTimer("dropped", "tick")
+ if err != nil || saved == nil || saved.Registration == "" || saved.ClaimOwner != owner.ClaimOwner {
+ t.Fatalf("worker lost claim mode: %+v, %v", saved, err)
+ }
+ timer.ClaimOwnership = owner
+ timer.ClaimOwner = "other"
+ if err := r.UpsertTimer(timer); !errors.Is(err, registry.ErrClaimOwnership) {
+ t.Fatalf("worker accepted conflicting claim: %v", err)
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ info, err := os.Stat(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ gotUID, gotGID, ok := fileOwner(info)
+ if !ok || gotUID != uid || gotGID != gid {
+ t.Fatalf("registry owner = %d:%d, want %d:%d", gotUID, gotGID, uid, gid)
+ }
+}
+
+func TestRegistryChildRefusesRootCredentials(t *testing.T) {
+ if _, err := openRegistryChild(filepath.Join(t.TempDir(), "registry.db"), 0, 0); err == nil {
+ t.Fatal("root worker accepted")
+ }
+}
+
+func TestRegistryChildCommandEnvironment(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "true")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "must-not-reach-worker")
+ cmd := exec.Command(filepath.Join(t.TempDir(), "missing-executable"))
+ if _, err := startRegistryChild(cmd, "unused", 65534, 65534); err == nil {
+ t.Fatal("missing worker started")
+ }
+ if len(cmd.Env) != 1 || cmd.Env[0] != "CARLOS_HOST_CLAIM=true" {
+ t.Fatalf("worker environment = %q", cmd.Env)
+ }
+ if cmd.Dir != "/" || cmd.SysProcAttr.Credential.Uid == 0 || cmd.SysProcAttr.Credential.NoSetGroups || len(cmd.SysProcAttr.Credential.Groups) != 0 {
+ t.Fatalf("unsafe worker command: %+v", cmd)
+ }
+}
diff --git a/internal/host/registryowner.go b/internal/host/registryowner.go
index 954979ab2..0bd7a53a9 100644
--- a/internal/host/registryowner.go
+++ b/internal/host/registryowner.go
@@ -13,45 +13,7 @@ import (
"github.com/carlosframework/platform/internal/registry"
)
-// The registry is written by TWO users. `carlos agent` (carlos-edge.service)
-// runs as serviceUser; every adoption-shaped pass — carlos-adopt,
-// carlos-stage, carlos-restart, carlos-config-sync — runs as root, because a
-// pass writes /opt/carlos and cycles units via systemctl. Both open
-// /data/registry.db, and SQLite creates the database and its -wal/-shm
-// sidecars owned by whoever got there first.
-//
-// On an established box the edge got there first, the files are carlos-owned,
-// and root writing to them is fine. On a FRESH box it is a race that root can
-// win: `carlos bootstrap` arms adopt's .path and .timer before the edge has
-// ever started, so adopt can create the registry as root and leave the edge
-// unable to start at all:
-//
-// agent: open registry /data/registry.db: attempt to write a readonly database (8)
-//
-// systemd gives up after five restarts in three seconds, so the box then sits
-// failed with no further attempts. All three region boxes brought up on
-// 2026-08-31 landed exactly there (discussion #8).
-//
-// Two fences, because one is not enough:
-//
-// 1. bootstrap creates the file itself, before the units that could race for
-// it are armed (see bootstrap.go) — so root never gets to create it.
-// 2. every root pass converges the ownership afterwards, below. Chowning by
-// hand once is NOT enough on a live box and the report says so: the armed
-// path/timer units fire again and re-create the sidecars as root in
-// between. Two of the three boxes needed the chown twice.
-
-// openServiceRegistry opens the registry for a root-side pass and converges
-// its ownership in the same breath.
-//
-// One function rather than two statements, because the two must not drift.
-// Opening the database is what CREATES the -wal/-shm, so the converge has to
-// follow the open — and `carlos adopt -restart-only` returns from RunAdopt a
-// few lines after it. A converge written as a separate statement that ever
-// slid below that early return would silently stop self-healing the restart
-// path, which is the pass that fires most often on a live box (carlos-restart
-// .path is per-request). Callers get the registry only through here, so there
-// is no ordering left to get wrong.
+// Root passes keep SQLite in a child running as the service user.
func openServiceRegistry(path string) (*registry.Registry, error) {
return openRegistryConverging(os.Geteuid(), serviceUser, path)
}
@@ -59,22 +21,31 @@ func openServiceRegistry(path string) (*registry.Registry, error) {
// openRegistryConverging is openServiceRegistry with the two facts it reads
// from the process passed in; see chownRegistryFiles.
func openRegistryConverging(euid int, svcUser, path string) (*registry.Registry, error) {
- reg, err := registry.Open(path)
+ if euid != 0 {
+ return registry.Open(path)
+ }
+ u, err := user.Lookup(svcUser)
if err != nil {
return nil, err
}
- chownRegistryFiles(euid, svcUser, path)
- return reg, nil
+ uid, err := strconv.Atoi(u.Uid)
+ if err != nil {
+ return nil, err
+ }
+ gid, err := strconv.Atoi(u.Gid)
+ if err != nil {
+ return nil, err
+ }
+ if uid <= 0 || gid < 0 {
+ return nil, fmt.Errorf("registry: invalid service credentials")
+ }
+ if err := chownRegistryFiles(euid, svcUser, path); err != nil {
+ return nil, err
+ }
+ return openRegistryChild(path, uid, gid)
}
-// chownRegistryToService leaves the registry and its WAL sidecars owned by
-// serviceUser. It is a no-op unless this process is root: a pass running as
-// the service user already creates everything with the right owner.
-//
-// Failures are logged, never fatal. An adoption pass that can read the
-// registry has real work to do — releases to stage, units to cycle — and
-// refusing to do it because a chown failed would turn a cosmetic problem into
-// an outage. The next pass tries again.
+// chownRegistryToService repairs old root-owned files through checked descriptors.
func chownRegistryToService(path string) {
// Geteuid reports -1 on platforms without the concept (the CLI
// cross-compiles to windows), which takes the same early exit.
@@ -85,45 +56,39 @@ func chownRegistryToService(path string) {
// effective uid and the service user's name — made arguments, so a test can
// drive the root arm, and both of its failure arms, from a test binary that
// is not root on a box that has no carlos user.
-func chownRegistryFiles(euid int, svcUser, path string) {
+func chownRegistryFiles(euid int, svcUser, path string) error {
if euid != 0 {
- return
+ return nil
}
u, err := user.Lookup(svcUser)
if err != nil {
log.Printf("adopt: registry ownership: no %s user: %v", svcUser, err)
- return
+ return err
}
uid, err := strconv.Atoi(u.Uid)
if err != nil {
- return
+ return err
}
gid, err := strconv.Atoi(u.Gid)
if err != nil {
- return
+ return err
}
+ var refused error
for _, p := range registryFiles(path) {
changed, err := ownRegistryFile(p, uid, gid)
if errors.Is(err, os.ErrNotExist) {
- // The -wal/-shm sidecars only exist while some connection is
- // open. Their absence is the normal state, not a problem.
continue
}
if err != nil {
- // Logged and stepped over, never returned: the files after this
- // one may well be fixable, and none of it is worth the pass. A
- // symlink or hard link here is refused, never followed.
- log.Printf("adopt: chown %s to %s: %v", p, svcUser, err)
+ log.Printf("adopt: ALERT: chown %s to %s: %v", p, svcUser, err)
+ refused = errors.Join(refused, err)
continue
}
- if !changed {
- continue
+ if changed {
+ log.Printf("adopt: %s was not owned by %s — fixed; the edge runs as %s and could not have written it", p, svcUser, svcUser)
}
- // Worth a line every time it fires: on a healthy box it never does,
- // so a run of these is the box telling you something armed root is
- // creating the registry ahead of the edge.
- log.Printf("adopt: %s was not owned by %s — fixed; the edge runs as %s and could not have written it", p, svcUser, svcUser)
}
+ return refused
}
// registryFiles is the database and the two sidecars SQLite keeps beside it
diff --git a/internal/host/registryowner_chown_unix.go b/internal/host/registryowner_chown_unix.go
index 0d7acec33..772e0d249 100644
--- a/internal/host/registryowner_chown_unix.go
+++ b/internal/host/registryowner_chown_unix.go
@@ -6,6 +6,10 @@ import (
"errors"
"fmt"
"os"
+ "path/filepath"
+ "strings"
+
+ "golang.org/x/sys/unix"
"syscall"
)
@@ -22,7 +26,16 @@ var errRegistryFileShape = errors.New("not a plain single-link regular file")
// it. A missing file reports os.ErrNotExist; changed is false when the file
// was already owned as wanted.
func ownRegistryFile(path string, uid, gid int) (changed bool, err error) {
- f, err := os.OpenFile(path, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK|syscall.O_CLOEXEC, 0)
+ parent, name, err := openRegistryParent(path)
+ if err != nil {
+ return false, err
+ }
+ defer parent.Close()
+ fd, err := openRegistryDescriptor(int(parent.Fd()), name)
+ var f *os.File
+ if err == nil {
+ f = os.NewFile(uintptr(fd), path)
+ }
if err != nil {
if errors.Is(err, syscall.ELOOP) {
return false, fmt.Errorf("%s: %w", path, errRegistryFileShape)
@@ -41,8 +54,32 @@ func ownRegistryFile(path string, uid, gid int) (changed bool, err error) {
if int(st.Uid) == uid && int(st.Gid) == gid {
return false, nil
}
- if err := f.Chown(uid, gid); err != nil {
+ if err := chownRegistryDescriptor(f, uid, gid); err != nil {
return false, err
}
return true, nil
}
+
+// Walk each parent without following symlinks, keeping the final directory pinned.
+func openRegistryParent(path string) (*os.File, string, error) {
+ absolute, err := filepath.Abs(path)
+ if err != nil {
+ return nil, "", err
+ }
+ parent, err := os.Open("/")
+ if err != nil {
+ return nil, "", err
+ }
+ for _, part := range strings.Split(strings.TrimPrefix(filepath.Dir(absolute), "/"), "/") {
+ if part == "" {
+ continue
+ }
+ fd, err := unix.Openat(int(parent.Fd()), part, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
+ parent.Close()
+ if err != nil {
+ return nil, "", err
+ }
+ parent = os.NewFile(uintptr(fd), part)
+ }
+ return parent, filepath.Base(absolute), nil
+}
diff --git a/internal/host/registryowner_descriptor_linux.go b/internal/host/registryowner_descriptor_linux.go
new file mode 100644
index 000000000..d98cd1b39
--- /dev/null
+++ b/internal/host/registryowner_descriptor_linux.go
@@ -0,0 +1,16 @@
+//go:build linux
+
+package host
+
+import (
+ "golang.org/x/sys/unix"
+ "os"
+)
+
+func openRegistryDescriptor(parent int, name string) (int, error) {
+ return unix.Openat(parent, name, unix.O_PATH|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
+}
+
+func chownRegistryDescriptor(f *os.File, uid, gid int) error {
+ return unix.Fchownat(int(f.Fd()), "", uid, gid, unix.AT_EMPTY_PATH|unix.AT_SYMLINK_NOFOLLOW)
+}
diff --git a/internal/host/registryowner_descriptor_unix.go b/internal/host/registryowner_descriptor_unix.go
new file mode 100644
index 000000000..6392ef788
--- /dev/null
+++ b/internal/host/registryowner_descriptor_unix.go
@@ -0,0 +1,16 @@
+//go:build unix && !linux
+
+package host
+
+import (
+ "golang.org/x/sys/unix"
+ "os"
+)
+
+func openRegistryDescriptor(parent int, name string) (int, error) {
+ return unix.Openat(parent, name, unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_NONBLOCK|unix.O_CLOEXEC, 0)
+}
+
+func chownRegistryDescriptor(f *os.File, uid, gid int) error {
+ return f.Chown(uid, gid)
+}
diff --git a/internal/host/registryowner_test.go b/internal/host/registryowner_test.go
index 782c837c9..3013160f3 100644
--- a/internal/host/registryowner_test.go
+++ b/internal/host/registryowner_test.go
@@ -271,27 +271,19 @@ func TestRegistryOpenHintSilentOnOtherErrors(t *testing.T) {
}
}
-// TestOpenServiceRegistryConvergesOwnership pins the two halves as one
-// operation. `carlos adopt -restart-only` returns from RunAdopt within a few
-// lines of the open, so a converge that ever slid below that early return
-// would silently stop self-healing the pass that fires most often. Opening
-// through this helper is what makes the ordering unloseable — the test drives
-// the root arm with a service user that cannot resolve, so the one thing it
-// observes is that the open reached the converge at all.
-func TestOpenServiceRegistryConvergesOwnership(t *testing.T) {
+func TestOpenServiceRegistryRequiresServiceUser(t *testing.T) {
path := filepath.Join(t.TempDir(), "registry.db")
- logged := captureLog(t)
const noSuchUser = "carlos-no-such-service-user-8f21"
reg, err := openRegistryConverging(0, noSuchUser, path)
- if err != nil {
- t.Fatalf("open: %v", err)
+ if reg != nil {
+ reg.Close()
+ t.Fatal("root opened a registry without a service identity")
}
- defer reg.Close()
- if _, err := reg.List(); err != nil {
- t.Fatalf("the helper returned an unusable registry: %v", err)
+ if err == nil {
+ t.Fatal("missing service user was accepted")
}
- if !strings.Contains(logged(), noSuchUser) {
- t.Errorf("opening the registry did not run the ownership converge; log was:\n%s", logged())
+ if _, statErr := os.Lstat(path); !errors.Is(statErr, os.ErrNotExist) {
+ t.Fatalf("root created the registry before resolving credentials: %v", statErr)
}
}
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index 144c37edd..8fb21b3e8 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -146,10 +146,7 @@ func openSystemStore(ctx context.Context) (release.Store, ed25519.PublicKey, err
return store, pub, nil
}
-// openSystemRegistry is the production openRegistry: the box's registry,
-// opened as the root passes open it so any WAL files the open creates end
-// up owned by the service user, and wrapped with its path so a refusal
-// that names an open error says which file.
+// openSystemRegistry keeps root's SQLite access in the service-user child.
func openSystemRegistry(dataDir string) (*registry.Registry, error) {
path := filepath.Join(dataDir, "registry.db")
reg, err := openServiceRegistry(path)
diff --git a/internal/registry/claimhold.go b/internal/registry/claimhold.go
index 43eb8097f..5f8f87b5f 100644
--- a/internal/registry/claimhold.go
+++ b/internal/registry/claimhold.go
@@ -6,6 +6,11 @@ import "context"
// The caller supplies phase authority; this transaction binds the exact row
// and observed operation, and does not change the failed-host projection.
func (r *Registry) SetClaimHold(ctx context.Context, host string, owner ClaimOwnership, expected, next string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "SetClaimHold", Host: host, Owner: owner, Expected: expected, Next: next}, nil)
+ return err
+ }
+
if !owner.Valid() || host == "" {
return ErrClaimOwnership
}
diff --git a/internal/registry/claimrows.go b/internal/registry/claimrows.go
index f581b8901..3051588b8 100644
--- a/internal/registry/claimrows.go
+++ b/internal/registry/claimrows.go
@@ -53,6 +53,11 @@ func (r *Registry) migrateClaimOwnership() error {
// AdoptClaimRoute is called only by the complete-census backfill coordinator.
// The full observed row is compared inside the write transaction.
func (r *Registry) AdoptClaimRoute(ctx context.Context, observed Route, owner ClaimOwnership) error {
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "AdoptClaimRoute", Observed: observed, Owner: owner}, nil)
+ return err
+ }
+
if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) {
return ErrClaimOwnership
}
@@ -88,6 +93,14 @@ func (r *Registry) AdoptClaimRoute(ctx context.Context, observed Route, owner Cl
// A missing or replacement row is successful only when this journal's committed
// archive proves the predecessor was already removed by an earlier attempt.
func (r *Registry) RetireClaimRows(ctx context.Context, journal string, targets []ClaimRow, archive func([]ClaimRowArchive) error) error {
+ if archive == nil {
+ return ErrClaimOwnership
+ }
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "RetireClaimRows", Journal: journal, Targets: targets}, func(out remoteResponse) error { return archive(out.Archives) })
+ return err
+ }
+
if journal == "" || len(targets) == 0 || archive == nil {
return ErrClaimOwnership
}
@@ -189,6 +202,11 @@ func (r *Registry) RetireClaimRows(ctx context.Context, journal string, targets
}
func (r *Registry) AdoptClaimTimer(ctx context.Context, observed Timer, owner ClaimOwnership) error {
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "AdoptClaimTimer", ObservedValue: observed, Owner: owner}, nil)
+ return err
+ }
+
if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) || observed.Registration == "" || owner.ClaimRowID != observed.Registration {
return ErrClaimOwnership
}
diff --git a/internal/registry/claimtimers.go b/internal/registry/claimtimers.go
index 5c2a2e1b3..11464aded 100644
--- a/internal/registry/claimtimers.go
+++ b/internal/registry/claimtimers.go
@@ -12,6 +12,11 @@ var ErrTimerChanged = errors.New("registry: timer registration changed")
// UpsertTimerRegistration compares the captured registration and returns the
// resulting registration in the same transaction. Empty means absent at read.
func (r *Registry) UpsertTimerRegistration(t Timer, expected string) (*Timer, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "UpsertTimerRegistration", T: t, Expected: expected}, nil)
+ return out.Timer, err
+ }
+
tx, err := r.sql.Begin()
if err != nil {
return nil, err
@@ -64,6 +69,11 @@ func (r *Registry) UpsertTimerRegistration(t Timer, expected string) (*Timer, er
}
func (r *Registry) SetTimerStatusRegistration(host, name, registration, status string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetTimerStatusRegistration", Host: host, Name: name, Registration: registration, Status: status}, nil)
+ return err
+ }
+
if registration == "" {
return ErrTimerChanged
}
@@ -72,6 +82,11 @@ func (r *Registry) SetTimerStatusRegistration(host, name, registration, status s
}
func (r *Registry) FireTimerNowRegistration(host, name, registration string, now int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "FireTimerNowRegistration", Host: host, Name: name, Registration: registration, Now: now}, nil)
+ return err
+ }
+
if registration == "" {
return ErrTimerChanged
}
diff --git a/internal/registry/coldhandover.go b/internal/registry/coldhandover.go
index 6e832076a..a47e33f4d 100644
--- a/internal/registry/coldhandover.go
+++ b/internal/registry/coldhandover.go
@@ -15,6 +15,11 @@ type ColdRows struct {
}
func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string, old ColdRows, next ColdRows) error {
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "RestoreClaimRows", Id: id, Retirement: retirement, Old: old, NextValue: next}, nil)
+ return err
+ }
+
if id == "" || retirement == "" || !old.Route.ClaimOwnership.Valid() || !next.Route.ClaimOwnership.Valid() || old.Route.Host != next.Route.Host || old.Route.ClaimIncarnation != next.Route.ClaimIncarnation || next.Route.ClaimStamp <= old.Route.ClaimStamp || len(old.Timers) != len(next.Timers) {
return ErrClaimOwnership
}
diff --git a/internal/registry/fleets.go b/internal/registry/fleets.go
index 3593ddfef..31c67ad3b 100644
--- a/internal/registry/fleets.go
+++ b/internal/registry/fleets.go
@@ -151,6 +151,11 @@ func (r *Registry) freshFleetsConn() (*sql.DB, error) {
// UpsertFleet inserts or updates a fleet's account/bucket linkage by name.
func (r *Registry) UpsertFleet(f Fleet) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "UpsertFleet", F: f}, nil)
+ return err
+ }
+
if !ValidTunnelName(f.Name) {
return fmt.Errorf("invalid fleet name %q", f.Name)
}
@@ -176,6 +181,11 @@ ON CONFLICT(name) DO UPDATE SET
// ON CONFLICT DO UPDATE would. created reports which caller won; a losing
// call makes no change to the existing row at all — not even a rev bump.
func (r *Registry) CreateFleet(f Fleet) (created bool, err error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "CreateFleet", F: f}, nil)
+ return out.Changed, err
+ }
+
if !ValidTunnelName(f.Name) {
return false, fmt.Errorf("invalid fleet name %q", f.Name)
}
@@ -199,6 +209,11 @@ ON CONFLICT(name) DO NOTHING`,
// GetFleet returns the fleet named name, or (nil, nil) if there is none.
func (r *Registry) GetFleet(name string) (*Fleet, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "GetFleet", Name: name}, nil)
+ return out.Fleet, err
+ }
+
row := r.sql.QueryRow(`SELECT name, account, bucket, created_at FROM fleets WHERE name = ?`, name)
var f Fleet
err := row.Scan(&f.Name, &f.Account, &f.Bucket, &f.CreatedAt)
@@ -213,6 +228,11 @@ func (r *Registry) GetFleet(name string) (*Fleet, error) {
// ListFleets returns every fleet, ordered by name, over a fresh connection.
func (r *Registry) ListFleets() ([]Fleet, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ListFleets"}, nil)
+ return out.Fleets, err
+ }
+
db, err := r.freshFleetsConn()
if err != nil {
return nil, err
@@ -260,6 +280,11 @@ func (r *Registry) ListFleets() ([]Fleet, error) {
// string matching, is how a caller tells this ordinary conflict apart from
// an actual failure below.
func (r *Registry) AddFleetBox(b FleetBox) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "AddFleetBox", B: b}, nil)
+ return err
+ }
+
if !ValidTunnelName(b.Fleet) {
return fmt.Errorf("invalid fleet name %q", b.Fleet)
}
@@ -322,6 +347,11 @@ WHERE fleet_boxes.status <> 'active'`,
// rev: a sha256 token_hash is always 64 characters, so the rotation alone
// changes no length Fingerprint could otherwise key off.
func (r *Registry) SetFleetBoxToken(fleet, label, tokenHash string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetFleetBoxToken", Fleet: fleet, Label: label, TokenHash: tokenHash}, nil)
+ return err
+ }
+
if !ValidTunnelName(fleet) {
return fmt.Errorf("invalid fleet name %q", fleet)
}
@@ -348,6 +378,11 @@ func (r *Registry) SetFleetBoxToken(fleet, label, tokenHash string) error {
// SetFleetBoxTags mirrors SetFleetBoxToken: same rev bump, same
// RowsAffected guard for an unknown (fleet, label).
func (r *Registry) SetFleetBoxTags(fleet, label, tags string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetFleetBoxTags", Fleet: fleet, Label: label, Tags: tags}, nil)
+ return err
+ }
+
if !ValidTunnelName(fleet) {
return fmt.Errorf("invalid fleet name %q", fleet)
}
@@ -377,6 +412,11 @@ func (r *Registry) SetFleetBoxTags(fleet, label, tags string) error {
// off status = 'active', so a typo'd status here would make a live box
// invisible to it.
func (r *Registry) SetFleetBoxStatus(fleet, label, status string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetFleetBoxStatus", Fleet: fleet, Label: label, Status: status}, nil)
+ return err
+ }
+
if !ValidTunnelName(fleet) {
return fmt.Errorf("invalid fleet name %q", fleet)
}
@@ -403,6 +443,11 @@ func (r *Registry) SetFleetBoxStatus(fleet, label, status string) error {
// ListFleetBoxes returns every box registered to fleet, ordered by label,
// over a fresh connection (same rationale as ListFleets).
func (r *Registry) ListFleetBoxes(fleet string) ([]FleetBox, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ListFleetBoxes", Fleet: fleet}, nil)
+ return out.FleetBoxes, err
+ }
+
db, err := r.freshFleetsConn()
if err != nil {
return nil, err
@@ -457,6 +502,11 @@ func PlacementTarget(fleet string) string {
// cross-process poll List/Fingerprint exist for (the edge holds a long-lived
// Registry handle and reloads this whenever Fingerprint moves).
func (r *Registry) FleetChannelAuth() (map[string]string, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "FleetChannelAuth"}, nil)
+ return out.StringMap, err
+ }
+
db, err := r.freshFleetsConn()
if err != nil {
return nil, err
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index a4b410bf2..f2c98fe24 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -24,8 +24,9 @@ import (
)
type Registry struct {
- sql *sql.DB
- path string
+ remote *remoteClient
+ sql *sql.DB
+ path string
}
// Route is one host a deployment serves: an app instance the platform
@@ -467,7 +468,12 @@ func (r *Registry) Path() string { return r.path }
// console) deliberately never call this — they hold one handle for the
// process lifetime — but a test that rebuilds a registry file from scratch
// must be able to let go of it first.
-func (r *Registry) Close() error { return r.sql.Close() }
+func (r *Registry) Close() error {
+ if r.remote != nil {
+ return r.remote.close()
+ }
+ return r.sql.Close()
+}
// migrate is additive (CREATE TABLE IF NOT EXISTS), so deploying new code
// over an existing registry is always safe. New columns go in the ALTER list —
@@ -709,6 +715,11 @@ type Account struct {
// which is what makes account identity survive a registry rebuild and be
// identical on every box in a deployment.
func (r *Registry) PutAccount(id int64, sqid, name string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "PutAccount", IdValue: id, Sqid: sqid, Name: name}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`INSERT INTO accounts (id, sqid, name) VALUES (?, ?, ?)
ON CONFLICT(id) DO UPDATE SET sqid = excluded.sqid, name = excluded.name`, id, sqid, name)
return err
@@ -727,6 +738,11 @@ ON CONFLICT(id) DO UPDATE SET sqid = excluded.sqid, name = excluded.name`, id, s
// runs the other way — domains imports registry — and a constant is a cheaper
// thing to keep in step than a cycle is to break.
func (r *Registry) DeleteAccount(sqid string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeleteAccount", Sqid: sqid}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM accounts WHERE sqid = ? AND id != ?`, sqid, opsAccountID)
return err
}
@@ -736,6 +752,11 @@ const opsAccountID = 1
// AccountBySqid resolves an account's public identity to its id and name, or
// (0, "", nil) when absent — "no such account" is an answer, not an error.
func (r *Registry) AccountBySqid(sqid string) (int64, string, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "AccountBySqid", Sqid: sqid}, nil)
+ return out.Number64, out.Text, err
+ }
+
var id int64
var name string
err := r.sql.QueryRow(`SELECT id, name FROM accounts WHERE sqid = ?`, sqid).Scan(&id, &name)
@@ -751,6 +772,11 @@ func (r *Registry) AccountBySqid(sqid string) (int64, string, error) {
// Accounts returns every materialized account, ordered by id (so ops is
// always first).
func (r *Registry) Accounts() ([]Account, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "Accounts"}, nil)
+ return out.Accounts, err
+ }
+
rows, err := r.sql.Query(`SELECT id, sqid, name FROM accounts ORDER BY id`)
if err != nil {
return nil, err
@@ -806,6 +832,11 @@ func scanRoute(scan func(...any) error) (Route, error) {
// matching, never retriggers. (Observed live in titogo, 2026-07-25.) List is
// reload-rare, so the ~1ms open cost is nothing.
func (r *Registry) List() ([]Route, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "List"}, nil)
+ return out.Routes, err
+ }
+
db, err := sql.Open("sqlite", r.path)
if err != nil {
return nil, err
@@ -833,6 +864,11 @@ func (r *Registry) List() ([]Route, error) {
// Get returns the route for host, or (nil, nil) if there is none.
func (r *Registry) Get(host string) (*Route, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "Get", Host: host}, nil)
+ return out.Route, err
+ }
+
row := r.sql.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host = ?`, host)
t, err := scanRoute(row.Scan)
if err == sql.ErrNoRows {
@@ -847,6 +883,11 @@ func (r *Registry) Get(host string) (*Route, error) {
// Delete removes a route by host (data-safe: it stops routing the host but
// never touches its DB).
func (r *Registry) Delete(host string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "Delete", Host: host}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM routes WHERE host = ?`, host)
return err
}
@@ -975,6 +1016,11 @@ func (r *Registry) Delete(host string) error {
// Route53 latency records — a real incident this repo caught in review before
// it shipped, not a hypothetical.
func (r *Registry) Upsert(t Route) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "Upsert", TValue: t}, nil)
+ return err
+ }
+
if t.CreatedAt == 0 {
t.CreatedAt = time.Now().Unix()
}
@@ -1074,6 +1120,11 @@ func IsHold(status string) bool { return HoldStatuses[status] }
// 13's rollback hold rule relies on it); SetHold itself does not, because the
// marker lives outside this table.
func (r *Registry) SetHold(host, status, op string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetHold", Host: host, Status: status, Op: op}, nil)
+ return err
+ }
+
if !IsHold(status) || op == "" {
return fmt.Errorf("registry: %q with op %q is not a hold", status, op)
}
@@ -1104,6 +1155,11 @@ func (r *Registry) SetHold(host, status, op string) error {
// As with SetHold, every caller touches the stage-requested marker after a
// write that landed.
func (r *Registry) SetHoldIf(host, status, op, expectOp string) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "SetHoldIf", Host: host, Status: status, Op: op, ExpectOp: expectOp}, nil)
+ return out.Changed, err
+ }
+
if !IsHold(status) || op == "" {
return false, fmt.Errorf("registry: %q with op %q is not a hold", status, op)
}
@@ -1131,6 +1187,11 @@ func (r *Registry) SetHoldIf(host, status, op, expectOp string) (bool, error) {
// wraps it under WithHostHoldLock and touches the stage-requested marker
// after a hand-over that landed, which this package cannot reach.
func (r *Registry) HandHold(host, status, toOp, fromOp string) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "HandHold", Host: host, Status: status, ToOp: toOp, FromOp: fromOp}, nil)
+ return out.Changed, err
+ }
+
if !IsHold(status) || toOp == "" || fromOp == "" || toOp == fromOp {
return false, fmt.Errorf("registry: hand %s from %q to %q as %q is not a hand-over between two ops' holds", host, fromOp, toOp, status)
}
@@ -1146,6 +1207,11 @@ func (r *Registry) HandHold(host, status, toOp, fromOp string) (bool, error) {
// under the op that holds it. false means nothing matched (a late write from a
// finished operation, an unheld row, or an empty op).
func (r *Registry) ReleaseHold(host, op string) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ReleaseHold", Host: host, Op: op}, nil)
+ return out.Changed, err
+ }
+
if op == "" {
return false, nil
}
@@ -1161,6 +1227,11 @@ func (r *Registry) ReleaseHold(host, op string) (bool, error) {
// deletion at finalization (moves spec §7.9 step 3). A row held by another
// op, or not held, is not deleted and the answer is false.
func (r *Registry) DeleteIfHold(host, op string) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DeleteIfHold", Host: host, Op: op}, nil)
+ return out.Changed, err
+ }
+
if op == "" {
return false, nil
}
@@ -1178,6 +1249,11 @@ func (r *Registry) DeleteIfHold(host, op string) (bool, error) {
// and nothing was deleted, timers included, so a timers clear never runs
// ahead of the row it belongs to or lands on a successor's rows.
func (r *Registry) DeleteIfHoldWithTimers(host, op string) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DeleteIfHoldWithTimers", Host: host, Op: op}, nil)
+ return out.Changed, err
+ }
+
if op == "" {
return false, nil
}
@@ -1206,6 +1282,11 @@ func (r *Registry) DeleteIfHoldWithTimers(host, op string) (bool, error) {
// host. Over a fresh connection, like List, so a reader in another process
// sees the latest commit.
func (r *Registry) HoldRows() ([]Route, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "HoldRows"}, nil)
+ return out.Routes, err
+ }
+
db, err := sql.Open("sqlite", r.path)
if err != nil {
return nil, err
@@ -1240,6 +1321,11 @@ func (r *Registry) HoldRows() ([]Route, error) {
// writers give up after busy_timeout. Nothing is written: the transaction is
// rolled back. fn's error is returned.
func (r *Registry) WithHoldsFrozen(fn func(held []Route) error) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "WithHoldsFrozen"}, func(out remoteResponse) error { return fn(out.Held) })
+ return err
+ }
+
ctx := context.Background()
conn, err := r.sql.Conn(ctx)
if err != nil {
@@ -1276,6 +1362,11 @@ func (r *Registry) WithHoldsFrozen(fn func(held []Route) error) error {
// read-modify-write Upsert racing a deploy's SetVersions (or the activator's
// SetNextDue) would clobber those columns.
func (r *Registry) SetAccount(host, account string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetAccount", Host: host, Account: account}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET account = ? WHERE host = ?`, account, host)
return err
}
@@ -1287,6 +1378,11 @@ func (r *Registry) SetAccount(host, account string) error {
// route` can report an unknown host rather than silently doing nothing — this
// is the one setter a human types a hostname into directly.
func (r *Registry) SetChannel(host, channel string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetChannel", Host: host, Channel: channel}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET channel = ? WHERE host = ?`, channel, host)
if err != nil {
return err
@@ -1308,6 +1404,11 @@ func (r *Registry) SetChannel(host, channel string) error {
// the edge's served set and the activator's managed set in one write, which
// is what quiesces the child ahead of the final mirror round.
func (r *Registry) SetStatus(host, status string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetStatus", Host: host, Status: status}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET status = ? WHERE host = ?`, status, host)
if err != nil {
return err
@@ -1344,6 +1445,11 @@ func (r *Registry) SetStatus(host, status string) error {
// lost race is a refusal that names the current value; the operator re-runs
// against reality.
func (r *Registry) SetCaps(host, from, to string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetCaps", Host: host, From: from, To: to}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET caps = ? WHERE host = ? AND caps = ?`, to, host, from)
if err != nil {
return err
@@ -1395,6 +1501,11 @@ func capsOrNone(caps string) string {
// replicator streams. A lost race is a refusal naming the current value, and
// the next pass re-decides against reality.
func (r *Registry) SetDBPath(host, from, to string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetDBPath", Host: host, From: from, To: to}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET db_path = ? WHERE host = ? AND db_path = ?`, to, host, from)
if err != nil {
return err
@@ -1429,6 +1540,11 @@ var ErrPathsChanged = errors.New("registry: instance paths changed underneath th
// swapped together because a row with one flat path and one nested path is
// neither layout, and every consumer would refuse it.
func (r *Registry) SetInstancePaths(host, fromDB, fromSock, toDB, toSock string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetInstancePaths", Host: host, FromDB: fromDB, FromSock: fromSock, ToDB: toDB, ToSock: toSock}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET db_path = ?, socket = ? WHERE host = ? AND db_path = ? AND socket = ?`, toDB, toSock, host, fromDB, fromSock)
if err != nil {
return err
@@ -1470,6 +1586,11 @@ func (r *Registry) SetInstancePaths(host, fromDB, fromSock, toDB, toSock string)
// (SetDBPath's): it is one operator, at one step of a documented migration,
// naming the value they want.
func (r *Registry) SetBacking(host, backing string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetBacking", Host: host, Backing: backing}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET backing = ? WHERE host = ?`, backing, host)
if err != nil {
return err
@@ -1510,6 +1631,11 @@ func (r *Registry) SetBacking(host, backing string) error {
// nothing, i.e. as the very bug this column closes. "" is accepted, because
// clearing the binding is a legitimate write (see ValidUnitName).
func (r *Registry) SetUnit(host, unit string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetUnit", Host: host, Unit: unit}, nil)
+ return err
+ }
+
if !ValidUnitName(unit) {
return fmt.Errorf("unit %q: want a systemd service unit name (e.g. console.service) or \"\" to clear", unit)
}
@@ -1543,6 +1669,11 @@ func (r *Registry) SetUnit(host, unit string) error {
// to. It reaches here only from `carlos route --replication -`, never from
// a full-row Upsert, which preserves the column instead.
func (r *Registry) SetReplication(host, class string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetReplication", Host: host, Class: class}, nil)
+ return err
+ }
+
if !ValidReplication(class) {
return fmt.Errorf("replication %q: want one of %s, or \"\" for the default", class, strings.Join(ReplicationClasses(), ", "))
}
@@ -1594,6 +1725,11 @@ func (r *Registry) SetReplication(host, class string) error {
// no-op on a host that does not exist would hide a reconcile pass writing to
// the wrong row.
func (r *Registry) SetUpstreams(host, addr, addrs, health string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetUpstreams", Host: host, Addr: addr, Addrs: addrs, Health: health}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET addr = ?, addrs = ?, health = ? WHERE host = ?`,
addr, addrs, health, host)
if err != nil {
@@ -1627,6 +1763,11 @@ func (r *Registry) SetUpstreams(host, addr, addrs, health string) error {
// reconcile pass that has just read the row must notice it vanishing
// underneath, not silently do nothing.
func (r *Registry) SetSteering(host, steering string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetSteering", Host: host, Steering: steering}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET steering = ? WHERE host = ?`, steering, host)
if err != nil {
return err
@@ -1652,6 +1793,11 @@ func (r *Registry) SetSteering(host, steering string) error {
// SetChannel's contract: the reconciler must notice the row vanishing
// underneath rather than silently do nothing.
func (r *Registry) SetReplicas(host, replicas string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetReplicas", Host: host, Replicas: replicas}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET replicas = ? WHERE host = ?`, replicas, host)
if err != nil {
return err
@@ -1675,6 +1821,11 @@ func (r *Registry) SetReplicas(host, replicas string) error {
// contract verbatim, because a reconcile pass that has just read the row
// must notice it vanishing underneath, not silently do nothing.
func (r *Registry) SetReplica(host string, replica bool) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetReplica", Host: host, Replica: replica}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET replica = ? WHERE host = ?`, b2iDB(replica), host)
if err != nil {
return err
@@ -1703,6 +1854,11 @@ func b2iDB(b bool) int {
// gate by design and so could not clear one). Errors on an unmatched host,
// SetBacking's contract.
func (r *Registry) SetGate(host, hash string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetGate", Host: host, Hash: hash}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET gate = ? WHERE host = ?`, hash, host)
if err != nil {
return err
@@ -1728,6 +1884,11 @@ func (r *Registry) SetGate(host, hash string) error {
// must never be left half-moved. Same "error on an unmatched host" contract
// as SetChannel, but any single failure rolls the whole batch back.
func (r *Registry) SetChannels(hosts []string, channel string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetChannels", Hosts: hosts, Channel: channel}, nil)
+ return err
+ }
+
if len(hosts) == 0 {
return nil
}
@@ -1771,6 +1932,11 @@ func (r *Registry) SetChannels(hosts []string, channel string) error {
// CLI's "-environment -" sentinel resolves to that before calling here, so
// this function itself has no notion of "clear" beyond writing "".
func (r *Registry) SetEnvironment(host, env string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetEnvironment", Host: host, Env: env}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET environment = ? WHERE host = ?`, env, host)
if err != nil {
return err
@@ -1793,6 +1959,11 @@ func (r *Registry) SetEnvironment(host, env string) error {
// CLI's "-tunnel -" sentinel resolves to that before calling here, so this
// function itself has no notion of "clear" beyond writing "".
func (r *Registry) SetTunnel(host, tunnel string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetTunnel", Host: host, Tunnel: tunnel}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET tunnel = ? WHERE host = ?`, tunnel, host)
if err != nil {
return err
@@ -1813,6 +1984,11 @@ func (r *Registry) SetTunnel(host, tunnel string) error {
// placement pass's door, distinct from SetPlacement (which writes the
// placement TARGET, not the box currently fulfilling it).
func (r *Registry) SetFleetBox(host, label string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetFleetBox", Host: host, Label: label}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET fleet_box = ? WHERE host = ?`, label, host)
if err != nil {
return err
@@ -1841,6 +2017,11 @@ func (r *Registry) SetFleetBox(host, label string) error {
// human or a placement pass typing a hostname here should get a refusal on a
// typo, not a silent no-op.
func (r *Registry) SetPlacement(host, target string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetPlacement", Host: host, Target: target}, nil)
+ return err
+ }
+
if target != "" {
if _, ok := FleetPlacement(target); !ok {
return fmt.Errorf("invalid placement target %q (want \"\" or \"fleet/<name>\")", target)
@@ -1867,6 +2048,11 @@ func (r *Registry) SetPlacement(host, target string) error {
// every route's existence-rail check to completion FIRST, and only once
// every route has cleared it does it call this to write.
func (r *Registry) SetEnvironments(hosts []string, env string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetEnvironments", Hosts: hosts, Env: env}, nil)
+ return err
+ }
+
if len(hosts) == 0 {
return nil
}
@@ -1911,6 +2097,11 @@ func (r *Registry) SetEnvironments(hosts []string, env string) error {
// single-column siblings; the caller (runRoute) has already run both rails'
// checks to completion before this is called.
func (r *Registry) SetChannelsAndEnvironments(hosts []string, channel, env string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetChannelsAndEnvironments", Hosts: hosts, Channel: channel, Env: env}, nil)
+ return err
+ }
+
if len(hosts) == 0 {
return nil
}
@@ -1949,6 +2140,11 @@ func (r *Registry) SetChannelsAndEnvironments(hosts []string, channel, env strin
// SetDNSStatus stamps a route's DNS health verdict + check time, narrowly
// (same rationale as SetAccount; this lands every few minutes on every host).
func (r *Registry) SetDNSStatus(host, status string, checkedAt int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetDNSStatus", Host: host, Status: status, CheckedAt: checkedAt}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET dns_status = ?, dns_checked_at = ? WHERE host = ?`, status, checkedAt, host)
return err
}
@@ -1956,6 +2152,11 @@ func (r *Registry) SetDNSStatus(host, status string, checkedAt int64) error {
// SetCertExpiresAt stamps a route's cached-cert NotAfter, narrowly. 0 =
// unknown (no cert in the cache, or one we couldn't parse).
func (r *Registry) SetCertExpiresAt(host string, at int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetCertExpiresAt", Host: host, At: at}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET cert_expires_at = ? WHERE host = ?`, at, host)
return err
}
@@ -1964,6 +2165,11 @@ func (r *Registry) SetCertExpiresAt(host string, at int64) error {
// so a slow deploy's read-modify-write can never clobber concurrent writes to
// other columns.
func (r *Registry) SetVersions(host, ver, prev string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetVersions", Host: host, Ver: ver, Prev: prev}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET version = ?, prev_version = ? WHERE host = ?`, ver, prev, host)
return err
}
@@ -1994,6 +2200,11 @@ func (r *Registry) SetVersions(host, ver, prev string) error {
// naming the rung again costs one command; runRouteHibernate prints what was
// dropped so it is a fact they are told, not one they discover.
func (r *Registry) SetHibernate(host string, on bool) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetHibernate", Host: host, On: on}, nil)
+ return err
+ }
+
if !on {
_, err := r.sql.Exec(`UPDATE routes SET hibernate = 0, replication = '' WHERE host = ?`, host)
return err
@@ -2005,6 +2216,11 @@ func (r *Registry) SetHibernate(host string, on bool) error {
// SetNextDue records the earliest server-side scheduled work for a host, so
// the activator can wake a sleeping instance for its own schedule. 0 = none.
func (r *Registry) SetNextDue(host string, due int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetNextDue", Host: host, Due: due}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET next_due = ? WHERE host = ?`, due, host)
return err
}
@@ -2029,6 +2245,11 @@ func (r *Registry) SetNextDue(host string, due int64) error {
// next_due = ? is what keeps consumption from eating it. A row that moved
// under us is simply left alone.
func (r *Registry) ClearNextDueIf(host string, due int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "ClearNextDueIf", Host: host, Due: due}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET next_due = 0 WHERE host = ? AND next_due = ?`, host, due)
return err
}
@@ -2043,6 +2264,11 @@ func (r *Registry) ClearNextDueIf(host string, due int64) error {
// it (host's workersToStart): only "ready", or the legacy "" some pre-status
// rows carry — a row still provisioning, or draining, must not be spawned.
func (r *Registry) AlwaysOnWorkers() ([]Route, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "AlwaysOnWorkers"}, nil)
+ return out.Routes, err
+ }
+
rows, err := r.sql.Query(`SELECT ` + routeCols + ` FROM routes
WHERE kind = 'sidecar' AND hibernate = 0 AND (status = '' OR status = 'ready') ORDER BY host`)
if err != nil {
@@ -2067,6 +2293,11 @@ WHERE kind = 'sidecar' AND hibernate = 0 AND (status = '' OR status = 'ready') O
// instance-provisioning reconcile pass re-stamps this every tick and must
// notice a row that vanished under it rather than silently doing nothing.
func (r *Registry) SetSeed(host, seed string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetSeed", Host: host, Seed: seed}, nil)
+ return err
+ }
+
res, err := r.sql.Exec(`UPDATE routes SET seed = ? WHERE host = ?`, seed, host)
if err != nil {
return err
@@ -2084,6 +2315,11 @@ func (r *Registry) SetSeed(host, seed string) error {
// SetArgs repoints a sidecar's argv — the narrow UPDATE the sidecar
// reconcile pass uses when the app's manifest changed.
func (r *Registry) SetArgs(host, args string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetArgs", Host: host, Args: args}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE routes SET args = ? WHERE host = ?`, args, host)
return err
}
@@ -2093,6 +2329,11 @@ func (r *Registry) SetArgs(host, args string) error {
// and same reason as SetArgs: the manifest is the truth, this is the copy
// the sweeper reads on every tick.
func (r *Registry) SetWorkerLimits(host string, idle, maxRun int) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetWorkerLimits", Host: host, Idle: idle, MaxRun: maxRun}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(
`UPDATE routes SET idle_seconds = ?, max_run_seconds = ? WHERE host = ?`,
idle, maxRun, host)
@@ -2111,6 +2352,11 @@ func (r *Registry) SetWorkerLimits(host string, idle, maxRun int) error {
// adoption step (provision.Reconcile, design §4.2) should ever set it, and
// it does so only after matching the row's shape against the record.
func (r *Registry) SetProvisioned(host string, on bool) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetProvisioned", Host: host, On: on}, nil)
+ return err
+ }
+
v := 0
if on {
v = 1
@@ -2136,6 +2382,11 @@ func (r *Registry) SetProvisioned(host string, on bool) error {
// reconcile pass converges this column from the instance record and must
// notice a row that vanished under it rather than silently doing nothing.
func (r *Registry) SetStatusNamed(host string, on bool) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetStatusNamed", Host: host, On: on}, nil)
+ return err
+ }
+
v := 0
if on {
v = 1
@@ -2159,6 +2410,11 @@ func (r *Registry) SetStatusNamed(host string, on bool) error {
// are included because the due-time sweep is their ONLY timer wake source:
// nothing can dial a sidecar, so no request will ever wake one.
func (r *Registry) DueInstances(now int64) ([]Route, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DueInstances", Now: now}, nil)
+ return out.Routes, err
+ }
+
rows, err := r.sql.Query(`SELECT `+routeCols+` FROM routes
WHERE kind IN ('instance','sidecar') AND next_due > 0 AND next_due <= ? ORDER BY host`, now)
if err != nil {
@@ -2198,6 +2454,11 @@ type RouteAlias struct {
// ListAliases returns every route alias, ordered by alias.
func (r *Registry) ListAliases() ([]RouteAlias, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ListAliases"}, nil)
+ return out.Aliases, err
+ }
+
rows, err := r.sql.Query(`SELECT alias, host, account, app, dns_status,
dns_checked_at, cert_expires_at, created_at FROM route_aliases ORDER BY alias`)
if err != nil {
@@ -2224,6 +2485,11 @@ dns_checked_at, cert_expires_at, created_at FROM route_aliases ORDER BY alias`)
// leaving those columns out of the SET list rather than a CASE guard, since
// every writer of this table already knows it must not carry stamps).
func (r *Registry) UpsertAlias(a RouteAlias) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "UpsertAlias", A: a}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`
INSERT INTO route_aliases (alias, host, account, app, dns_status, dns_checked_at, cert_expires_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
@@ -2235,6 +2501,11 @@ ON CONFLICT(alias) DO UPDATE SET host=excluded.host, account=excluded.account, a
// DeleteAlias removes an alias by its hostname. Idempotent: an alias already
// gone is a success, same contract as Delete on routes.
func (r *Registry) DeleteAlias(alias string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeleteAlias", Alias: alias}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM route_aliases WHERE alias = ?`, alias)
return err
}
@@ -2243,6 +2514,11 @@ func (r *Registry) DeleteAlias(alias string) error {
// narrowly (same rationale as SetDNSStatus on routes: it lands on its own
// cadence and must not race a concurrent UpsertAlias repoint).
func (r *Registry) SetAliasDNSStatus(alias, status string, at int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetAliasDNSStatus", Alias: alias, Status: status, At: at}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE route_aliases SET dns_status = ?, dns_checked_at = ? WHERE alias = ?`, status, at, alias)
return err
}
@@ -2250,6 +2526,11 @@ func (r *Registry) SetAliasDNSStatus(alias, status string, at int64) error {
// SetAliasCertExpiresAt stamps an alias's cached-cert NotAfter, narrowly.
// 0 = unknown (same rationale as SetCertExpiresAt on routes).
func (r *Registry) SetAliasCertExpiresAt(alias string, at int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetAliasCertExpiresAt", Alias: alias, At: at}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE route_aliases SET cert_expires_at = ? WHERE alias = ?`, at, alias)
return err
}
@@ -2268,6 +2549,11 @@ type WildcardDomain struct {
// ListWildcardDomains returns every wildcard domain, ordered by host.
func (r *Registry) ListWildcardDomains() ([]WildcardDomain, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ListWildcardDomains"}, nil)
+ return out.Wildcards, err
+ }
+
rows, err := r.sql.Query(`SELECT host, account, app, created_at FROM wildcard_domains ORDER BY host`)
if err != nil {
return nil, err
@@ -2286,6 +2572,11 @@ func (r *Registry) ListWildcardDomains() ([]WildcardDomain, error) {
// UpsertWildcardDomain inserts or updates a row by host.
func (r *Registry) UpsertWildcardDomain(w WildcardDomain) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "UpsertWildcardDomain", W: w}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`
INSERT INTO wildcard_domains (host, account, app, created_at)
VALUES (?, ?, ?, ?)
@@ -2297,6 +2588,11 @@ ON CONFLICT(host) DO UPDATE SET account=excluded.account, app=excluded.app, crea
// DeleteWildcardDomain removes host's row. Idempotent: a host already gone
// is a success, same contract as DeleteAlias.
func (r *Registry) DeleteWildcardDomain(host string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeleteWildcardDomain", Host: host}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM wildcard_domains WHERE host = ?`, host)
return err
}
@@ -2323,6 +2619,11 @@ type RoutePath struct {
// ListPaths returns every path route, ordered by host then prefix.
func (r *Registry) ListPaths() ([]RoutePath, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "ListPaths"}, nil)
+ return out.Paths, err
+ }
+
rows, err := r.sql.Query(`SELECT host, prefix, target, account, app
FROM route_paths ORDER BY host, prefix`)
if err != nil {
@@ -2344,6 +2645,11 @@ FROM route_paths ORDER BY host, prefix`)
// Unlike UpsertAlias there are no stamp columns to preserve — nothing on this
// table is maintained on a separate cadence — so every column is overwritten.
func (r *Registry) UpsertPath(p RoutePath) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "UpsertPath", P: p}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`
INSERT INTO route_paths (host, prefix, target, account, app)
VALUES (?, ?, ?, ?, ?)
@@ -2357,6 +2663,11 @@ ON CONFLICT(host, prefix) DO UPDATE SET
// success, the same contract as DeleteAlias — the reconcile pass deletes
// unconditionally from its own view of what is no longer desired.
func (r *Registry) DeletePath(host, prefix string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeletePath", Host: host, Prefix: prefix}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM route_paths WHERE host = ? AND prefix = ?`, host, prefix)
return err
}
@@ -2449,6 +2760,11 @@ func (r *Registry) DeletePath(host, prefix string) error {
// never a missed change — the failure mode worth engineering against is the
// false NEGATIVE.
func (r *Registry) Fingerprint() (string, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "Fingerprint"}, nil)
+ return out.Text, err
+ }
+
db, err := sql.Open("sqlite", r.path)
if err != nil {
return "", err
@@ -2679,6 +2995,11 @@ var snapshotClaimCols = map[string]bool{
}
func (r *Registry) SnapshotFingerprint() (string, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "SnapshotFingerprint"}, nil)
+ return out.Text, err
+ }
+
db, err := sql.Open("sqlite", r.path)
if err != nil {
return "", err
@@ -2813,6 +3134,11 @@ FROM fleet_boxes ORDER BY fleet, label`)
// containing a backslash gets VACUUM INTO writing to a different on-disk
// path than the one os.Stat just checked, silently defeating the guard.
func (r *Registry) SnapshotTo(path string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SnapshotTo", Path: path}, nil)
+ return err
+ }
+
if _, err := os.Stat(path); err == nil {
return fmt.Errorf("registry: snapshot target already exists: %s", path)
} else if !os.IsNotExist(err) {
diff --git a/internal/registry/remote.go b/internal/registry/remote.go
new file mode 100644
index 000000000..4a400c961
--- /dev/null
+++ b/internal/registry/remote.go
@@ -0,0 +1,243 @@
+package registry
+
+import (
+ "context"
+ "database/sql"
+ "encoding/gob"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "sync"
+ "time"
+)
+
+var errRemoteMethod = errors.New("registry: unsupported remote operation")
+
+var remoteErrors = []error{os.ErrNotExist, os.ErrPermission, sql.ErrNoRows, ErrClaimOwnership, ErrTimerChanged, ErrTimerKind, ErrPathsChanged, ErrFleetBoxActive, context.Canceled, context.DeadlineExceeded}
+
+func responseError(out remoteResponse) error {
+ if !out.Failed {
+ return nil
+ }
+ if out.ErrorKind > 0 && out.ErrorKind <= len(remoteErrors) {
+ sentinel := remoteErrors[out.ErrorKind-1]
+ if out.Error == sentinel.Error() {
+ return sentinel
+ }
+ return fmt.Errorf("%s: %w", out.Error, sentinel)
+ }
+ return errors.New(out.Error)
+}
+
+func setResponseError(out *remoteResponse, err error) {
+ if err == nil {
+ return
+ }
+ out.Failed = true
+ out.Error = err.Error()
+ for i, sentinel := range remoteErrors {
+ if errors.Is(err, sentinel) {
+ out.ErrorKind = i + 1
+ break
+ }
+ }
+}
+
+type remoteClient struct {
+ mu sync.Mutex
+ enc *gob.Encoder
+ dec *gob.Decoder
+ input io.WriteCloser
+ output io.ReadCloser
+ wait func() error
+ closed bool
+}
+
+// OpenRemote connects to an already privilege-dropped registry worker.
+func OpenRemote(path string, input io.WriteCloser, output io.ReadCloser, wait func() error) (*Registry, error) {
+ c := &remoteClient{enc: gob.NewEncoder(input), dec: gob.NewDecoder(output), input: input, output: output, wait: wait}
+ var ready remoteResponse
+ err := c.dec.Decode(&ready)
+ if err == nil {
+ err = responseError(ready)
+ }
+ if err != nil {
+ c.stop()
+ return nil, err
+ }
+ return &Registry{path: path, remote: c}, nil
+}
+
+func (c *remoteClient) stop() error {
+ if c.closed {
+ return nil
+ }
+ c.closed = true
+ c.input.Close()
+ c.output.Close()
+ if c.wait != nil {
+ return c.wait()
+ }
+ return nil
+}
+
+func (c *remoteClient) close() error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.stop()
+}
+
+func (c *remoteClient) call(ctx context.Context, q remoteRequest, callback func(remoteResponse) error) (out remoteResponse, err error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if c.closed {
+ return out, os.ErrClosed
+ }
+ if ctx != nil {
+ if err = ctx.Err(); err != nil {
+ return
+ }
+ if deadline, ok := ctx.Deadline(); ok {
+ q.Deadline = deadline.UnixNano()
+ }
+ }
+ // Closing the pipe cancels the worker context, including an in-flight SQL call.
+ if ctx != nil {
+ stopped := make(chan struct{})
+ stop := context.AfterFunc(ctx, func() { c.input.Close(); c.output.Close(); close(stopped) })
+ defer func() {
+ if !stop() {
+ <-stopped
+ }
+ }()
+ defer func() {
+ if ctx.Err() != nil {
+ err = ctx.Err()
+ c.stop()
+ }
+ }()
+ }
+ // A callback panic or a broken pipe must release any child transaction.
+ complete := false
+ defer func() {
+ if !complete {
+ c.stop()
+ }
+ }()
+ if err = c.enc.Encode(q); err != nil {
+ return
+ }
+ if err = c.dec.Decode(&out); err != nil {
+ return
+ }
+ var callbackErr error
+ if out.Callback != "" {
+ if callback == nil {
+ return out, errRemoteMethod
+ }
+ callbackErr = callback(out)
+ if ctx != nil && ctx.Err() != nil {
+ return out, ctx.Err()
+ }
+ ack := remoteRequest{Method: "callback"}
+ if callbackErr != nil {
+ ack.CallbackFailed = true
+ ack.CallbackError = callbackErr.Error()
+ }
+ if err = c.enc.Encode(ack); err != nil {
+ return
+ }
+ out = remoteResponse{}
+ if err = c.dec.Decode(&out); err != nil {
+ return
+ }
+ if out.Callback != "" {
+ return out, errRemoteMethod
+ }
+ }
+ complete = true
+ if callbackErr != nil {
+ return out, callbackErr
+ }
+ return out, responseError(out)
+}
+
+// ServeRemote owns SQLite in the worker. Callers must drop credentials first.
+func ServeRemote(path string, input io.Reader, output io.Writer) error {
+ if os.Geteuid() == 0 {
+ return errors.New("registry: remote worker refuses root")
+ }
+ enc, dec := gob.NewEncoder(output), gob.NewDecoder(input)
+ r, err := Open(path)
+ ready := remoteResponse{}
+ setResponseError(&ready, err)
+ if writeErr := enc.Encode(ready); writeErr != nil {
+ if r != nil {
+ r.Close()
+ }
+ return writeErr
+ }
+ if err != nil {
+ return err
+ }
+ defer r.Close()
+ workerCtx, stopWorker := context.WithCancel(context.Background())
+ defer stopWorker()
+ requests := make(chan remoteRequest)
+ readErrors := make(chan error, 1)
+ go func() {
+ defer close(requests)
+ defer stopWorker()
+ for {
+ var q remoteRequest
+ if err := dec.Decode(&q); err != nil {
+ readErrors <- err
+ return
+ }
+ select {
+ case requests <- q:
+ case <-workerCtx.Done():
+ return
+ }
+ }
+ }()
+ for q := range requests {
+ ctx := workerCtx
+ cancel := func() {}
+ if q.Deadline != 0 {
+ ctx, cancel = context.WithDeadline(ctx, time.Unix(0, q.Deadline))
+ }
+ out, callErr := r.dispatchRemote(ctx, q, func(event remoteResponse) error {
+ if err := enc.Encode(event); err != nil {
+ return err
+ }
+ ack, ok := <-requests
+ if !ok {
+ return workerCtx.Err()
+ }
+ if ack.Method != "callback" {
+ return errRemoteMethod
+ }
+ if ack.CallbackFailed {
+ return errors.New(ack.CallbackError)
+ }
+ return nil
+ })
+ cancel()
+ setResponseError(&out, callErr)
+ if err = enc.Encode(out); err != nil {
+ return err
+ }
+ }
+ select {
+ case err := <-readErrors:
+ if errors.Is(err, io.EOF) {
+ return nil
+ }
+ return err
+ default:
+ return workerCtx.Err()
+ }
+
+}
diff --git a/internal/registry/remote_dispatch.go b/internal/registry/remote_dispatch.go
new file mode 100644
index 000000000..bd3281455
--- /dev/null
+++ b/internal/registry/remote_dispatch.go
@@ -0,0 +1,318 @@
+package registry
+
+import "context"
+
+type remoteRequest struct {
+ Method string
+ Deadline int64
+ CallbackError string
+ CallbackFailed bool
+ Host string
+ Owner ClaimOwnership
+ Expected string
+ Next string
+ Observed Route
+ Journal string
+ Targets []ClaimRow
+ ObservedValue Timer
+ T Timer
+ Name string
+ Registration string
+ Status string
+ Now int64
+ Id string
+ Retirement string
+ Old ColdRows
+ NextValue ColdRows
+ F Fleet
+ B FleetBox
+ Fleet string
+ Label string
+ TokenHash string
+ Tags string
+ IdValue int64
+ Sqid string
+ TValue Route
+ Op string
+ ExpectOp string
+ ToOp string
+ FromOp string
+ Account string
+ Channel string
+ From string
+ To string
+ FromDB string
+ FromSock string
+ ToDB string
+ ToSock string
+ Backing string
+ Unit string
+ Class string
+ Addr string
+ Addrs string
+ Health string
+ Steering string
+ Replicas string
+ Replica bool
+ Hash string
+ Hosts []string
+ Env string
+ Tunnel string
+ Target string
+ CheckedAt int64
+ At int64
+ Ver string
+ Prev string
+ On bool
+ Due int64
+ Seed string
+ Args string
+ Idle int
+ MaxRun int
+ A RouteAlias
+ Alias string
+ W WildcardDomain
+ P RoutePath
+ Prefix string
+ Path string
+ Rows []Timer
+ Limit int
+ ExcludeHosts []string
+ Ms int64
+ Attempts int
+ NextAt int64
+ Kind string
+ Alive map[string]bool
+ DoneBefore int64
+ Keep map[string]bool
+}
+
+type remoteResponse struct {
+ Failed bool
+ Error string
+ ErrorKind int
+ Callback string
+ Held []Route
+ Archives []ClaimRowArchive
+ Timer *Timer
+ Changed bool
+ Fleet *Fleet
+ Fleets []Fleet
+ FleetBoxes []FleetBox
+ StringMap map[string]string
+ Number64 int64
+ Text string
+ Accounts []Account
+ Routes []Route
+ Route *Route
+ Aliases []RouteAlias
+ Wildcards []WildcardDomain
+ Paths []RoutePath
+ Timers []Timer
+ Number int
+ Times map[string]int64
+}
+
+func (r *Registry) dispatchRemote(ctx context.Context, q remoteRequest, callback func(remoteResponse) error) (out remoteResponse, err error) {
+ switch q.Method {
+ case "SetClaimHold":
+ err = r.SetClaimHold(ctx, q.Host, q.Owner, q.Expected, q.Next)
+ case "AdoptClaimRoute":
+ err = r.AdoptClaimRoute(ctx, q.Observed, q.Owner)
+ case "RetireClaimRows":
+ err = r.RetireClaimRows(ctx, q.Journal, q.Targets, func(archives []ClaimRowArchive) error {
+ return callback(remoteResponse{Callback: "archives", Archives: archives})
+ })
+ case "AdoptClaimTimer":
+ err = r.AdoptClaimTimer(ctx, q.ObservedValue, q.Owner)
+ case "UpsertTimerRegistration":
+ out.Timer, err = r.UpsertTimerRegistration(q.T, q.Expected)
+ case "SetTimerStatusRegistration":
+ err = r.SetTimerStatusRegistration(q.Host, q.Name, q.Registration, q.Status)
+ case "FireTimerNowRegistration":
+ err = r.FireTimerNowRegistration(q.Host, q.Name, q.Registration, q.Now)
+ case "RestoreClaimRows":
+ err = r.RestoreClaimRows(ctx, q.Id, q.Retirement, q.Old, q.NextValue)
+ case "UpsertFleet":
+ err = r.UpsertFleet(q.F)
+ case "CreateFleet":
+ out.Changed, err = r.CreateFleet(q.F)
+ case "GetFleet":
+ out.Fleet, err = r.GetFleet(q.Name)
+ case "ListFleets":
+ out.Fleets, err = r.ListFleets()
+ case "AddFleetBox":
+ err = r.AddFleetBox(q.B)
+ case "SetFleetBoxToken":
+ err = r.SetFleetBoxToken(q.Fleet, q.Label, q.TokenHash)
+ case "SetFleetBoxTags":
+ err = r.SetFleetBoxTags(q.Fleet, q.Label, q.Tags)
+ case "SetFleetBoxStatus":
+ err = r.SetFleetBoxStatus(q.Fleet, q.Label, q.Status)
+ case "ListFleetBoxes":
+ out.FleetBoxes, err = r.ListFleetBoxes(q.Fleet)
+ case "FleetChannelAuth":
+ out.StringMap, err = r.FleetChannelAuth()
+ case "PutAccount":
+ err = r.PutAccount(q.IdValue, q.Sqid, q.Name)
+ case "DeleteAccount":
+ err = r.DeleteAccount(q.Sqid)
+ case "AccountBySqid":
+ out.Number64, out.Text, err = r.AccountBySqid(q.Sqid)
+ case "Accounts":
+ out.Accounts, err = r.Accounts()
+ case "List":
+ out.Routes, err = r.List()
+ case "Get":
+ out.Route, err = r.Get(q.Host)
+ case "Delete":
+ err = r.Delete(q.Host)
+ case "Upsert":
+ err = r.Upsert(q.TValue)
+ case "SetHold":
+ err = r.SetHold(q.Host, q.Status, q.Op)
+ case "SetHoldIf":
+ out.Changed, err = r.SetHoldIf(q.Host, q.Status, q.Op, q.ExpectOp)
+ case "HandHold":
+ out.Changed, err = r.HandHold(q.Host, q.Status, q.ToOp, q.FromOp)
+ case "ReleaseHold":
+ out.Changed, err = r.ReleaseHold(q.Host, q.Op)
+ case "DeleteIfHold":
+ out.Changed, err = r.DeleteIfHold(q.Host, q.Op)
+ case "DeleteIfHoldWithTimers":
+ out.Changed, err = r.DeleteIfHoldWithTimers(q.Host, q.Op)
+ case "HoldRows":
+ out.Routes, err = r.HoldRows()
+ case "WithHoldsFrozen":
+ err = r.WithHoldsFrozen(func(held []Route) error { return callback(remoteResponse{Callback: "holds", Held: held}) })
+ case "SetAccount":
+ err = r.SetAccount(q.Host, q.Account)
+ case "SetChannel":
+ err = r.SetChannel(q.Host, q.Channel)
+ case "SetStatus":
+ err = r.SetStatus(q.Host, q.Status)
+ case "SetCaps":
+ err = r.SetCaps(q.Host, q.From, q.To)
+ case "SetDBPath":
+ err = r.SetDBPath(q.Host, q.From, q.To)
+ case "SetInstancePaths":
+ err = r.SetInstancePaths(q.Host, q.FromDB, q.FromSock, q.ToDB, q.ToSock)
+ case "SetBacking":
+ err = r.SetBacking(q.Host, q.Backing)
+ case "SetUnit":
+ err = r.SetUnit(q.Host, q.Unit)
+ case "SetReplication":
+ err = r.SetReplication(q.Host, q.Class)
+ case "SetUpstreams":
+ err = r.SetUpstreams(q.Host, q.Addr, q.Addrs, q.Health)
+ case "SetSteering":
+ err = r.SetSteering(q.Host, q.Steering)
+ case "SetReplicas":
+ err = r.SetReplicas(q.Host, q.Replicas)
+ case "SetReplica":
+ err = r.SetReplica(q.Host, q.Replica)
+ case "SetGate":
+ err = r.SetGate(q.Host, q.Hash)
+ case "SetChannels":
+ err = r.SetChannels(q.Hosts, q.Channel)
+ case "SetEnvironment":
+ err = r.SetEnvironment(q.Host, q.Env)
+ case "SetTunnel":
+ err = r.SetTunnel(q.Host, q.Tunnel)
+ case "SetFleetBox":
+ err = r.SetFleetBox(q.Host, q.Label)
+ case "SetPlacement":
+ err = r.SetPlacement(q.Host, q.Target)
+ case "SetEnvironments":
+ err = r.SetEnvironments(q.Hosts, q.Env)
+ case "SetChannelsAndEnvironments":
+ err = r.SetChannelsAndEnvironments(q.Hosts, q.Channel, q.Env)
+ case "SetDNSStatus":
+ err = r.SetDNSStatus(q.Host, q.Status, q.CheckedAt)
+ case "SetCertExpiresAt":
+ err = r.SetCertExpiresAt(q.Host, q.At)
+ case "SetVersions":
+ err = r.SetVersions(q.Host, q.Ver, q.Prev)
+ case "SetHibernate":
+ err = r.SetHibernate(q.Host, q.On)
+ case "SetNextDue":
+ err = r.SetNextDue(q.Host, q.Due)
+ case "ClearNextDueIf":
+ err = r.ClearNextDueIf(q.Host, q.Due)
+ case "AlwaysOnWorkers":
+ out.Routes, err = r.AlwaysOnWorkers()
+ case "SetSeed":
+ err = r.SetSeed(q.Host, q.Seed)
+ case "SetArgs":
+ err = r.SetArgs(q.Host, q.Args)
+ case "SetWorkerLimits":
+ err = r.SetWorkerLimits(q.Host, q.Idle, q.MaxRun)
+ case "SetProvisioned":
+ err = r.SetProvisioned(q.Host, q.On)
+ case "SetStatusNamed":
+ err = r.SetStatusNamed(q.Host, q.On)
+ case "DueInstances":
+ out.Routes, err = r.DueInstances(q.Now)
+ case "ListAliases":
+ out.Aliases, err = r.ListAliases()
+ case "UpsertAlias":
+ err = r.UpsertAlias(q.A)
+ case "DeleteAlias":
+ err = r.DeleteAlias(q.Alias)
+ case "SetAliasDNSStatus":
+ err = r.SetAliasDNSStatus(q.Alias, q.Status, q.At)
+ case "SetAliasCertExpiresAt":
+ err = r.SetAliasCertExpiresAt(q.Alias, q.At)
+ case "ListWildcardDomains":
+ out.Wildcards, err = r.ListWildcardDomains()
+ case "UpsertWildcardDomain":
+ err = r.UpsertWildcardDomain(q.W)
+ case "DeleteWildcardDomain":
+ err = r.DeleteWildcardDomain(q.Host)
+ case "ListPaths":
+ out.Paths, err = r.ListPaths()
+ case "UpsertPath":
+ err = r.UpsertPath(q.P)
+ case "DeletePath":
+ err = r.DeletePath(q.Host, q.Prefix)
+ case "Fingerprint":
+ out.Text, err = r.Fingerprint()
+ case "SnapshotFingerprint":
+ out.Text, err = r.SnapshotFingerprint()
+ case "SnapshotTo":
+ err = r.SnapshotTo(q.Path)
+ case "UpsertTimer":
+ err = r.UpsertTimer(q.T)
+ case "ReplaceTimers":
+ err = r.ReplaceTimers(q.Host, q.Rows)
+ case "GetTimer":
+ out.Timer, err = r.GetTimer(q.Host, q.Name)
+ case "DeleteTimer":
+ err = r.DeleteTimer(q.Host, q.Name)
+ case "DeleteTimerRegistration":
+ err = r.DeleteTimerRegistration(q.Host, q.Name, q.Registration)
+ case "TimersFor":
+ out.Timers, err = r.TimersFor(q.Host)
+ case "DueTimers":
+ out.Timers, err = r.DueTimers(q.Now, q.Limit, q.ExcludeHosts)
+ case "DueUnheldTimers":
+ out.Timers, err = r.DueUnheldTimers(q.Now, q.Limit, q.ExcludeHosts)
+ case "FireTimerNow":
+ err = r.FireTimerNow(q.Host, q.Name, q.Now)
+ case "MarkTimerFired":
+ out.Changed, err = r.MarkTimerFired(q.Host, q.Name, q.Registration, q.At, q.Status, q.Ms, q.Attempts, q.NextAt, q.Due)
+ case "SetTimerStatus":
+ err = r.SetTimerStatus(q.Host, q.Name, q.Status)
+ case "CountTimers":
+ out.Number, err = r.CountTimers(q.Host, q.Kind)
+ case "PruneTimers":
+ out.Number, err = r.PruneTimers(q.Alive, q.DoneBefore)
+ case "DeleteTimersExcept":
+ out.Number, err = r.DeleteTimersExcept(q.Host, q.Kind, q.Keep)
+ case "EarliestTimers":
+ out.Times, err = r.EarliestTimers()
+ default:
+ err = errRemoteMethod
+ }
+ return
+}
diff --git a/internal/registry/remote_test.go b/internal/registry/remote_test.go
new file mode 100644
index 000000000..e714d0eba
--- /dev/null
+++ b/internal/registry/remote_test.go
@@ -0,0 +1,177 @@
+package registry
+
+import (
+ "context"
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func remoteTestRegistry(t *testing.T) (*Registry, string) {
+ t.Helper()
+ if os.Geteuid() == 0 {
+ t.Skip("in-process transport fixture must not open service files as root; credential tests use a dropped child")
+ }
+ path := filepath.Join(t.TempDir(), "registry.db")
+ requestR, requestW := io.Pipe()
+ responseR, responseW := io.Pipe()
+ done := make(chan error, 1)
+ go func() {
+ err := ServeRemote(path, requestR, responseW)
+ requestR.Close()
+ responseW.Close()
+ done <- err
+ }()
+ r, err := OpenRemote(path, requestW, responseR, func() error { return <-done })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { r.Close() })
+ return r, path
+}
+
+func TestRemoteRegistryRoundTrip(t *testing.T) {
+ r, _ := remoteTestRegistry(t)
+ route := Route{Host: "remote.example.com", Kind: "instance", Status: "ready"}
+ if err := r.Upsert(route); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.SetVersions(route.Host, "v2", "v1"); err != nil {
+ t.Fatal(err)
+ }
+ got, err := r.Get(route.Host)
+ if err != nil || got == nil || got.Version != "v2" || got.PrevVersion != "v1" {
+ t.Fatalf("route = %+v, %v", got, err)
+ }
+ timer, err := r.UpsertTimerRegistration(Timer{Host: route.Host, Name: "tick", Kind: "cron", Spec: "* * * * *"}, "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ saved, err := r.GetTimer(route.Host, "tick")
+ if err != nil || saved == nil || saved.Registration == "" || saved.Registration != timer.Registration {
+ t.Fatalf("timer = %+v, %v", saved, err)
+ }
+ if _, err := r.GetTimer(route.Host, "missing"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("missing sentinel = %v", err)
+ }
+ if err := r.SetClaimHold(t.Context(), route.Host, ClaimOwnership{}, "", "op"); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("claim sentinel = %v", err)
+ }
+ if err := r.RetireClaimRows(t.Context(), "journal", nil, nil); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatalf("nil archive = %v", err)
+ }
+}
+
+func TestRemoteHoldsFrozenKeepsWriteLock(t *testing.T) {
+ r, path := remoteTestRegistry(t)
+ if err := r.Upsert(Route{Host: "held", Kind: "instance"}); err != nil {
+ t.Fatal(err)
+ }
+ local, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer local.Close()
+ finished := make(chan error, 1)
+ callbackErr := errors.New("parent refused activation")
+ err = r.WithHoldsFrozen(func(rows []Route) error {
+ go func() { finished <- local.SetHold("held", "moving", "op") }()
+ select {
+ case err := <-finished:
+ t.Fatalf("writer escaped frozen transaction: %v", err)
+ case <-time.After(60 * time.Millisecond):
+ }
+ return callbackErr
+ })
+ if err != callbackErr {
+ t.Fatalf("callback error = %v", err)
+ }
+ select {
+ case err := <-finished:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(6 * time.Second):
+ t.Fatal("writer remained locked after callback")
+ }
+}
+
+func TestRemoteRetirementArchiveAndCancellation(t *testing.T) {
+ for _, mode := range []string{"commit", "cancel", "empty-error"} {
+ t.Run(mode, func(t *testing.T) {
+ cancelRequest := mode == "cancel"
+ callbackError := errors.New("")
+ r, path := remoteTestRegistry(t)
+ owner := ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"}
+ if err := r.Upsert(Route{Host: "claim", Kind: "instance"}); err != nil {
+ t.Fatal(err)
+ }
+ row, err := r.Get("claim")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = r.AdoptClaimRoute(t.Context(), *row, owner); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ err = r.RetireClaimRows(ctx, "journal", []ClaimRow{{Kind: "route", Host: "claim", Ownership: owner}}, func(archives []ClaimRowArchive) error {
+ if len(archives) != 1 || archives[0].Row.Ownership != owner {
+ t.Fatalf("archive = %+v", archives)
+ }
+ if cancelRequest {
+ cancel()
+ <-ctx.Done()
+ }
+ if mode == "empty-error" {
+ return callbackError
+ }
+ return nil
+ })
+ if mode == "empty-error" && err != callbackError {
+ t.Fatalf("empty callback error = %v", err)
+ }
+ if cancelRequest && !errors.Is(err, context.Canceled) {
+ t.Fatalf("cancel = %v", err)
+ }
+ if mode == "commit" && err != nil {
+ t.Fatal(err)
+ }
+ local, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer local.Close()
+ current, err := local.Get("claim")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if (mode != "commit") != (current != nil) {
+ t.Fatalf("cancel=%v: remaining row = %+v", cancelRequest, current)
+ }
+ })
+ }
+}
+
+func TestRemoteCallbackPanicRollsBack(t *testing.T) {
+ r, path := remoteTestRegistry(t)
+ func() {
+ defer func() {
+ if recover() == nil {
+ t.Error("callback did not panic")
+ }
+ }()
+ r.WithHoldsFrozen(func([]Route) error { panic("parent panic") })
+ }()
+ local, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer local.Close()
+ if err := local.Upsert(Route{Host: "after-panic", Kind: "instance"}); err != nil {
+ t.Fatalf("lock remained held: %v", err)
+ }
+}
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index 8ef7bd7cb..bb9891c3a 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -52,6 +52,11 @@ func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
// UpsertTimer inserts t, or updates an existing row's spec/path. A changed spec/path or non-zero NextAt starts a new registration and resets attempts; unchanged declarations preserve delivery state.
func (r *Registry) UpsertTimer(t Timer) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "UpsertTimer", T: t}, nil)
+ return err
+ }
+
if !hostwrite.Enabled(os.Getenv) {
return r.upsertLegacyTimer(t)
}
@@ -119,6 +124,11 @@ AND timers.claim_stamp=excluded.claim_stamp AND timers.claim_owner=excluded.clai
// UpsertTimer is the wrong tool for that: it mints a fresh registration
// and resets the runner's bookkeeping on a rearm.
func (r *Registry) ReplaceTimers(host string, rows []Timer) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "ReplaceTimers", Host: host, Rows: rows}, nil)
+ return err
+ }
+
if !hostwrite.Enabled(os.Getenv) {
return r.replaceLegacyTimers(host, rows)
}
@@ -158,6 +168,11 @@ func (r *Registry) ReplaceTimers(host string, rows []Timer) error {
}
func (r *Registry) GetTimer(host, name string) (*Timer, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "GetTimer", Host: host, Name: name}, nil)
+ return out.Timer, err
+ }
+
t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name))
if errors.Is(err, sql.ErrNoRows) {
return nil, os.ErrNotExist
@@ -169,11 +184,21 @@ func (r *Registry) GetTimer(host, name string) (*Timer, error) {
}
func (r *Registry) DeleteTimer(host, name string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimer", Host: host, Name: name}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=?`, host, name)
return err
}
func (r *Registry) DeleteTimerRegistration(host, name, registration string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimerRegistration", Host: host, Name: name, Registration: registration}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=? AND registration=?`, host, name, registration)
return err
}
@@ -196,6 +221,11 @@ func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
}
func (r *Registry) TimersFor(host string) ([]Timer, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "TimersFor", Host: host}, nil)
+ return out.Timers, err
+ }
+
return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, host)
}
@@ -211,6 +241,11 @@ func (r *Registry) TimersFor(host string) ([]Timer, error) {
// where the LIMIT is applied, instead of after it. An empty (or nil) slice
// is no filter, which is the ordinary case: nothing in flight.
func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DueTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil)
+ return out.Timers, err
+ }
+
return r.dueTimers(now, limit, excludeHosts, false)
}
@@ -222,6 +257,11 @@ func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Tim
// returns it. The runner asks this only with its hold wiring armed;
// DueTimers stays main's query.
func (r *Registry) DueUnheldTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DueUnheldTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil)
+ return out.Timers, err
+ }
+
return r.dueTimers(now, limit, excludeHosts, true)
}
@@ -253,6 +293,11 @@ func (r *Registry) dueTimers(now int64, limit int, excludeHosts []string, skipHe
// delivered (a `carlos schedule run` is its own occurrence, not an early
// delivery of the next natural one).
func (r *Registry) FireTimerNow(host, name string, now int64) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "FireTimerNow", Host: host, Name: name, Now: now}, nil)
+ return err
+ }
+
var id [16]byte
if _, err := rand.Read(id[:]); err != nil {
return err
@@ -267,6 +312,11 @@ func (r *Registry) FireTimerNow(host, name string, now int64) error {
// when the row advances, and the row's current DueAt when next is a retry
// of the occurrence just attempted — the runner decides which.
func (r *Registry) MarkTimerFired(host, name, registration string, at int64, status string, ms int64, attempts int, next, due int64) (bool, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "MarkTimerFired", Host: host, Name: name, Registration: registration, At: at, Status: status, Ms: ms, Attempts: attempts, NextAt: next, Due: due}, nil)
+ return out.Changed, err
+ }
+
res, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=? AND registration=?`,
at, status, ms, attempts, next, due, host, name, registration)
if err != nil {
@@ -280,11 +330,21 @@ func (r *Registry) MarkTimerFired(host, name, registration string, at int64, sta
// it parked (schedule.StatusNeverFires); last_at stays as it was, so the
// status reads as a condition, not a fire.
func (r *Registry) SetTimerStatus(host, name, status string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetTimerStatus", Host: host, Name: name, Status: status}, nil)
+ return err
+ }
+
_, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=?`, status, host, name)
return err
}
func (r *Registry) CountTimers(host, kind string) (int, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "CountTimers", Host: host, Kind: kind}, nil)
+ return out.Number, err
+ }
+
var n int
err := r.sql.QueryRow(`SELECT COUNT(*) FROM timers WHERE host=? AND kind=?`, host, kind).Scan(&n)
return n, err
@@ -302,6 +362,11 @@ func (r *Registry) CountTimers(host, kind string) (int, error) {
// list is empty rather than hand the wipe over. A third caller should do the
// same, or say in its own comment why the wipe is what it wants.
func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "PruneTimers", Alive: alive, DoneBefore: doneBefore}, nil)
+ return out.Number, err
+ }
+
hosts := make([]any, 0, len(alive))
for h := range alive {
hosts = append(hosts, h)
@@ -326,6 +391,11 @@ func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, er
}
func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimersExcept", Host: host, Kind: kind, Keep: keep}, nil)
+ return out.Number, err
+ }
+
names := make([]any, 0, len(keep))
for n := range keep {
names = append(names, n)
@@ -357,6 +427,11 @@ func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (
// the runner has not yet dated — and is excluded here rather than by the
// caller, matching DueTimers' own `next_at > 0` gate.
func (r *Registry) EarliestTimers() (map[string]int64, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(nil, remoteRequest{Method: "EarliestTimers"}, nil)
+ return out.Times, err
+ }
+
rows, err := r.sql.Query(`SELECT host, MIN(next_at) FROM timers WHERE next_at > 0 GROUP BY host`)
if err != nil {
return nil, err
diff --git a/main.go b/main.go
index 97b4fa892..90efe073b 100644
--- a/main.go
+++ b/main.go
@@ -26,6 +26,12 @@ func main() {
os.Exit(2)
}
cmd, args := os.Args[1], os.Args[2:]
+ if cmd == "__registry-child" {
+ if err := host.RunRegistryChild(args); err != nil {
+ log.Fatal(err)
+ }
+ return
+ }
switch cmd {
// Asking for help is not an error: it prints on stdout and exits 0, so
// `carlos help | less` works. A bare `carlos` still fails (stderr, 2).