amadan

carlos / platform Public

host, hostbox, registry: flag-off identical to main (registry NORMAL, adopt installs only missing units, bootstrap ignores host.env syntax); live-rendered control unit on update; completed producers count as effects (Plan 6b review round 1)

Opus I1-I3, M1, M3-M6; Codex Important 1-2.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude fb3ffd989fb10e4e35a195707b9f8a145b153795 parent 3015e88
30 files changed, +766 −35
  • internal/host/bootstrap.go +2 −1
  • internal/host/claim_isolation_test.go +4 −2
  • internal/host/claimassembly.go +3 −1
  • internal/host/claimconstruction.go +8 −0
  • internal/host/claimreview_test.go +204 −0
  • internal/host/claimstate_linux.go +45 −0
  • internal/host/claimstate_linux_test.go +20 −0
  • internal/host/claimstate_other.go +2 −0
  • internal/host/claimstop_run.go +6 −3
  • internal/host/claimunit.go +4 −1
  • internal/host/claimunit_test.go +10 −2
  • internal/host/claimwakeassembly.go +2 −4
  • internal/host/systemapply.go +14 −0
  • internal/host/systemunits.go +1 −7
  • internal/hostbox/coldhandover.go +4 −1
  • internal/hostbox/coldhandover_test.go +1 −0
  • internal/hostbox/intended_recovery_test.go +22 −0
  • internal/hostbox/journalpath.go +22 −0
  • internal/hostbox/journalpath_test.go +42 −0
  • internal/hostbox/launchjournal.go +3 −0
  • internal/hostbox/launchjournal_test.go +27 −0
  • internal/hostbox/stopinventory_db.go +3 −0
  • internal/registry/claimhold.go +7 −2
  • internal/registry/claimreview_test.go +167 −0
  • internal/registry/claimrows.go +3 −6
  • internal/registry/claimrows_test.go +2 −0
  • internal/registry/claimtransaction.go +74 −0
  • internal/registry/coldhandover.go +1 −1
  • internal/registry/registry.go +1 −1
  • internal/registry/timers.go +62 −3

CI passed — run details

diff --git a/internal/host/bootstrap.go b/internal/host/bootstrap.go
index 1026dfd7d..4693ea852 100644
--- a/internal/host/bootstrap.go
+++ b/internal/host/bootstrap.go
@@ -374,7 +374,8 @@ func bootstrapWith(root string, consumers []string, planJSON bool) error {
}
claimSettings, err := readClaimUnitSettings(hostEnv)
if err != nil {
- return err
+ fmt.Fprintf(os.Stderr, "bootstrap: instance-control unit settings: %v\n", err)
+ claimSettings = claimUnitSettings{}
}
beforeControl, err := os.ReadFile(filepath.Join(root, "etc", "systemd", "system", claimControlUnit))
if err != nil && !errors.Is(err, os.ErrNotExist) {
diff --git a/internal/host/claim_isolation_test.go b/internal/host/claim_isolation_test.go
index 1653ed0d2..057b27e4d 100644
--- a/internal/host/claim_isolation_test.go
+++ b/internal/host/claim_isolation_test.go
@@ -27,12 +27,14 @@ func (in isolationInputs) Select(ctx context.Context, rt registry.Route, o hostc
}
func TestClaimWakeRefusesEveryUnisolatedShapeBeforeAdmission(t *testing.T) {
- for _, shape := range []string{"nested", "flat", "foreign", "systemd", "seed", "standalone-sidecar", "co-spawn"} {
+ for _, shape := range []string{"nested", "flat", "foreign", "systemd", "seed", "standalone-sidecar", "co-spawn", "no-roles"} {
t.Run(shape, func(t *testing.T) {
f := newRecoveryFixture(t)
c := f.controller("boot-1", "controller-1", 4101, "41")
// Install the concrete production preflight, not a fixture override.
- c.wake.Roles = &ClaimRoleAssembly{Manager: c.wake.Manager, Journal: f.journal}
+ if shape != "no-roles" {
+ c.wake.Roles = &ClaimRoleAssembly{Manager: c.wake.Manager, Journal: f.journal}
+ }
roles := []string{"main"}
route := registry.Route{Host: recoveryHost, Kind: "instance", Hibernate: true}
switch shape {
diff --git a/internal/host/claimassembly.go b/internal/host/claimassembly.go
index 8088cf60b..1400857fb 100644
--- a/internal/host/claimassembly.go
+++ b/internal/host/claimassembly.go
@@ -54,7 +54,9 @@ func claimEntryContext(ctx context.Context, getenv func(string) string, open fun
if open == nil {
return ctx, hostwrite.ErrUnavailable
}
- store, err := open(ctx)
+ startup, cancel := context.WithTimeout(ctx, 2*time.Minute)
+ defer cancel()
+ store, err := open(startup)
if err != nil {
return ctx, err
}
diff --git a/internal/host/claimconstruction.go b/internal/host/claimconstruction.go
index 59c5e1592..2cb2abc79 100644
--- a/internal/host/claimconstruction.go
+++ b/internal/host/claimconstruction.go
@@ -122,6 +122,8 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla
if owners.Local == nil || owners.Controllers == nil || owners.Data == nil || owners.Inputs == nil || owners.InputPolicy == nil || owners.MovesPolicy == nil || owners.Evidence == nil || owners.LeaseSet == nil || owners.View == nil || owners.Authorize == nil || owners.AdmissionEvidence == nil || owners.PublicationEvidence == nil {
return nil, hostwrite.ErrUnavailable
}
+ ctx, finishStartup := context.WithTimeout(ctx, 2*time.Minute)
+ defer finishStartup()
actor, err := owners.Identity.Verify(ctx)
if err != nil {
return nil, err
@@ -133,6 +135,9 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla
if err = prepareClaimStateDirectory(local); err != nil {
return nil, err
}
+ if err = prepareClaimStateDatabase(filepath.Join(local, "launches.db")); err != nil {
+ return nil, err
+ }
launches, err := hostbox.OpenLaunchJournal(filepath.Join(local, "launches.db"))
if err != nil {
return nil, err
@@ -142,6 +147,9 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla
resultErr = errors.Join(resultErr, launches.Close())
}
}()
+ if err = prepareClaimStateDatabase(filepath.Join(local, "inventory.db")); err != nil {
+ return nil, err
+ }
inventory, err := hostbox.OpenStopInventoryDB(filepath.Join(local, "inventory.db"))
if err != nil {
return nil, err
diff --git a/internal/host/claimreview_test.go b/internal/host/claimreview_test.go
new file mode 100644
index 000000000..1548d48ee
--- /dev/null
+++ b/internal/host/claimreview_test.go
@@ -0,0 +1,204 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestClaimReviewFlagOffAdoptPreservesDrift(t *testing.T) {
+ etc := t.TempDir()
+ dir := filepath.Join(etc, "systemd/system")
+ if err := os.MkdirAll(dir, 0755); err != nil {
+ t.Fatal(err)
+ }
+ for _, name := range systemUnitFiles {
+ b, err := unitsFS.ReadFile("units/" + name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if name == claimControlUnit {
+ b = append(b, []byte("\n# operator customization\n")...)
+ }
+ if err := os.WriteFile(filepath.Join(dir, name), b, 0644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ before, _ := os.ReadFile(filepath.Join(dir, claimControlUnit))
+ calls := 0
+ ctl := func(...string) (string, error) { calls++; return "", nil }
+ probe := func(args ...string) (string, error) {
+ if args[0] == "is-active" {
+ return "active", nil
+ }
+ return "enabled", nil
+ }
+ ensureSystemUnits(etc, filepath.Join(t.TempDir(), "lock"), ctl, probe, t.Logf)
+ after, err := os.ReadFile(filepath.Join(dir, claimControlUnit))
+ if err != nil || calls != 0 || !bytes.Equal(before, after) {
+ t.Fatalf("flag-off drift: calls=%d changed=%v err=%v", calls, !bytes.Equal(before, after), err)
+ }
+}
+
+func TestClaimReviewMalformedSettingsInstallMissingControl(t *testing.T) {
+ etc := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(etc, "carlos"), 0755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(etc, "carlos/host.env"), []byte("CARLOS_HOST_CLAIM='unterminated\n"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ ensureSystemUnits(etc, filepath.Join(t.TempDir(), "lock"), func(...string) (string, error) { return "", nil }, func(...string) (string, error) { return "inactive", nil }, t.Logf)
+ got, err := os.ReadFile(filepath.Join(etc, "systemd/system", claimControlUnit))
+ want, _ := unitsFS.ReadFile("units/" + claimControlUnit)
+ if err != nil || !bytes.Equal(got, want) {
+ t.Fatalf("missing control not restored to main bytes: %v", err)
+ }
+}
+
+func TestClaimReviewBootstrapToleratesHostEnvironment(t *testing.T) {
+ for _, body := range []string{"not-a-setting\n", "CARLOS_X=foo\\\n", "CARLOS_X=\"unterminated\n", "CARLOS_HOST_CLAIM='unterminated\n"} {
+ t.Run(body, func(t *testing.T) {
+ root := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(root, "etc/carlos"), 0755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(root, "etc/carlos/host.env"), []byte(body), 0600); err != nil {
+ t.Fatal(err)
+ }
+ if err := bootstrap(root, nil); err != nil {
+ t.Fatalf("flag-off bootstrap failed: %v", err)
+ }
+ got, err := os.ReadFile(filepath.Join(root, "etc/systemd/system", claimControlUnit))
+ want, _ := unitsFS.ReadFile("units/" + claimControlUnit)
+ if err != nil || !bytes.Equal(got, want) {
+ t.Fatalf("not main bytes: %v", err)
+ }
+ })
+ }
+}
+
+func TestClaimReviewSystemStageUsesLiveControlSettings(t *testing.T) {
+ for _, enabled := range []bool{false, true} {
+ t.Run(map[bool]string{false: "off", true: "on"}[enabled], func(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1.0.0")
+ if err := os.MkdirAll(filepath.Join(f.a.etcDir, "carlos"), 0755); err != nil {
+ t.Fatal(err)
+ }
+ flag := ""
+ if enabled {
+ flag = "1"
+ }
+ if err := os.WriteFile(filepath.Join(f.a.etcDir, "carlos/host.env"), []byte("CARLOS_HOST_CLAIM="+flag+"\n"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ legacy, _ := unitsFS.ReadFile("units/" + claimControlUnit)
+ want := claimUnitContent(claimControlUnit, legacy, claimUnitSettings{enabled: enabled, data: "/data", opt: "/opt/carlos"})
+ livePath := filepath.Join(f.a.etcDir, "systemd/system", claimControlUnit)
+ if err := os.WriteFile(livePath, want, 0644); err != nil {
+ t.Fatal(err)
+ }
+ run := f.a.ops.run
+ f.a.ops.run = func(ctx context.Context, limit time.Duration, name string, args ...string) (string, error) {
+ if len(args) > 2 && args[0] == "bootstrap" {
+ if err := bootstrapWith(args[2], nil, true); err != nil {
+ return "", err
+ }
+ b, err := json.Marshal(currentBootstrapPlan())
+ return string(b), err
+ }
+ return run(ctx, limit, name, args...)
+ }
+ stage := filepath.Join(t.TempDir(), "stage")
+ if _, _, err := f.a.stagePlan(t.Context(), "candidate", stage); err != nil {
+ t.Fatal(err)
+ }
+ stagedPath := filepath.Join(stage, "etc/systemd/system", claimControlUnit)
+ staged, err := os.ReadFile(stagedPath)
+ if err != nil || !bytes.Equal(staged, legacy) {
+ t.Fatalf("stage read live settings: %v", err)
+ }
+ marker := []byte("\n# candidate release setting\n")
+ want = append(want, marker...)
+ if err := os.WriteFile(stagedPath, append(staged, marker...), 0644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(livePath, want, 0644); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.a.installUnits(t.Context(), stage, nil); err != nil {
+ t.Fatal(err)
+ }
+ got, err := os.ReadFile(livePath)
+ if err != nil || !bytes.Equal(got, want) {
+ t.Errorf("live control rendering changed: %v", err)
+ }
+ if strings.Contains(strings.Join(f.calls, "\n"), "try-restart "+claimControlUnit) {
+ t.Errorf("unchanged control restarted: %v", f.calls)
+ }
+ })
+ }
+}
+
+func TestClaimReviewControlEntryKeepsServiceLifetime(t *testing.T) {
+ service, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ var startup context.Context
+ store := release.NewDirStore(t.TempDir())
+ ctx, err := claimEntryContext(service, func(string) string { return "1" }, func(call context.Context) (release.Store, error) {
+ startup = call
+ if _, bounded := call.Deadline(); !bounded {
+ t.Error("store startup has no deadline")
+ }
+ return store, nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, bounded := ctx.Deadline(); bounded {
+ t.Fatal("service inherited startup deadline")
+ }
+ if startup.Err() != context.Canceled {
+ t.Error("startup context not released")
+ }
+ if ctx.Err() != nil || claimStore(ctx) != store {
+ t.Fatal("startup cancellation lost service or store")
+ }
+ sentinel := errors.New("provider stop")
+ owner := &reviewClaimOwner{open: func(call context.Context) error {
+ if _, bounded := call.Deadline(); bounded {
+ t.Error("owners inherited startup deadline")
+ }
+ if call.Err() != nil {
+ t.Error("owner context canceled before shutdown")
+ }
+ cancel()
+ if call.Err() != context.Canceled {
+ t.Error("owners do not follow service shutdown")
+ }
+ return sentinel
+ }}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ _, err = OpenProductionClaimAssembly(ctx, store, ClaimRuntimeResources{Mode: "instance-control", Registry: reg, DataDir: t.TempDir(), OptDir: t.TempDir(), UnitsDir: t.TempDir()}, owner)
+ if !errors.Is(err, sentinel) {
+ t.Fatal(err)
+ }
+}
+
+type reviewClaimOwner struct{ open func(context.Context) error }
+
+func (p *reviewClaimOwner) OpenClaimOwners(ctx context.Context, _ release.Store, _ ClaimRuntimeResources) (ClaimOwnerBindings, error) {
+ return ClaimOwnerBindings{}, p.open(ctx)
+}
diff --git a/internal/host/claimstate_linux.go b/internal/host/claimstate_linux.go
index e6e7ff1ef..274af6192 100644
--- a/internal/host/claimstate_linux.go
+++ b/internal/host/claimstate_linux.go
@@ -98,3 +98,48 @@ func claimStateDatabaseOwner(path string) error {
}
return nil
}
+
+func prepareClaimStateDatabase(path string) error {
+ uid, gid, err := claimStateOwner()
+ if err != nil {
+ return err
+ }
+ return prepareClaimDatabaseOwner(path, uid, gid)
+}
+func prepareClaimDatabaseOwner(path string, uid, gid int) error {
+ for _, suffix := range []string{"", "-wal", "-shm"} {
+ name := path + suffix
+ fd, err := unix.Open(name, unix.O_RDWR|unix.O_CLOEXEC|unix.O_NOFOLLOW|unix.O_NONBLOCK, 0)
+ created := false
+ if errors.Is(err, unix.ENOENT) {
+ if suffix != "" {
+ continue
+ }
+ fd, err = unix.Open(name, unix.O_RDWR|unix.O_CLOEXEC|unix.O_NOFOLLOW|unix.O_CREAT|unix.O_EXCL, 0600)
+ created = err == nil
+ }
+ if err != nil {
+ return err
+ }
+ err = func() error {
+ defer unix.Close(fd)
+ var stat unix.Stat_t
+ if err := unix.Fstat(fd, &stat); err != nil {
+ return err
+ }
+ if stat.Mode&unix.S_IFMT != unix.S_IFREG || (!created && int(stat.Uid) != uid) {
+ return hostwrite.ErrUnavailable
+ }
+ if created && os.Geteuid() == 0 {
+ if err := unix.Fchown(fd, uid, gid); err != nil {
+ return err
+ }
+ }
+ return unix.Fchmod(fd, 0600)
+ }()
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
diff --git a/internal/host/claimstate_linux_test.go b/internal/host/claimstate_linux_test.go
index 9a340ab32..0aceeb6df 100644
--- a/internal/host/claimstate_linux_test.go
+++ b/internal/host/claimstate_linux_test.go
@@ -49,3 +49,23 @@ func TestClaimPrivateStateRefusesSymlinkAndForeignOwner(t *testing.T) {
}
}
}
+
+func TestClaimReviewDatabaseOwnerCheckedBeforeSQLite(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "journal.db")
+ if err := prepareClaimDatabaseOwner(path, os.Getuid(), os.Getgid()); err != nil {
+ t.Fatal(err)
+ }
+ if err := prepareClaimDatabaseOwner(path, os.Getuid()+1, os.Getgid()); err == nil {
+ t.Fatal("foreign journal owner accepted")
+ }
+ victim := filepath.Join(t.TempDir(), "victim")
+ if err := os.Symlink(victim, path+"-wal"); err != nil {
+ t.Fatal(err)
+ }
+ if err := prepareClaimDatabaseOwner(path, os.Getuid(), os.Getgid()); err == nil {
+ t.Fatal("planted WAL accepted")
+ }
+ if _, err := os.Lstat(victim); !os.IsNotExist(err) {
+ t.Fatalf("touched victim: %v", err)
+ }
+}
diff --git a/internal/host/claimstate_other.go b/internal/host/claimstate_other.go
index fdf78b973..b6c6780ba 100644
--- a/internal/host/claimstate_other.go
+++ b/internal/host/claimstate_other.go
@@ -6,3 +6,5 @@ import "github.com/carlosframework/platform/internal/hostwrite"
func prepareClaimStateDirectory(string) error { return hostwrite.ErrUnavailable }
func claimStateDatabaseOwner(string) error { return hostwrite.ErrUnavailable }
+
+func prepareClaimStateDatabase(string) error { return hostwrite.ErrUnavailable }
diff --git a/internal/host/claimstop_run.go b/internal/host/claimstop_run.go
index 205162a2b..ff3a2fcfd 100644
--- a/internal/host/claimstop_run.go
+++ b/internal/host/claimstop_run.go
@@ -13,7 +13,7 @@ import (
)
func RunInstanceControl(args []string) {
- ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+ ctx, cancel := context.WithCancel(context.Background())
defer cancel()
ctx, err := claimEntryContext(ctx, os.Getenv, openDeployStore)
if err != nil {
@@ -41,7 +41,9 @@ func RunInstanceControl(args []string) {
log.Printf("instance-control: host claims shutdown: %v", err)
}
}()
- handler, err := NewProductionRootClaimStop(ctx)
+ startup, finishStartup := context.WithTimeout(ctx, 2*time.Minute)
+ defer finishStartup()
+ handler, err := NewProductionRootClaimStop(startup)
if err != nil {
log.Fatal(err)
}
@@ -49,8 +51,9 @@ func RunInstanceControl(args []string) {
if !ok {
log.Fatal(ErrClaimStopUnavailable)
}
- if err := recovery.RecoverPending(ctx); err != nil {
+ if err := recovery.RecoverPending(startup); err != nil {
log.Fatal(err)
}
+ finishStartup()
instancectl.RunWithClaim(args, handler)
}
diff --git a/internal/host/claimunit.go b/internal/host/claimunit.go
index d56d4c2c0..5d84859ff 100644
--- a/internal/host/claimunit.go
+++ b/internal/host/claimunit.go
@@ -47,7 +47,10 @@ func readClaimUnitSettings(path string) (claimUnitSettings, error) {
key, v, ok := strings.Cut(line, "=")
key = strings.TrimSpace(key)
v = strings.TrimSpace(v)
- if !ok || strings.HasSuffix(v, `\`) || (len(v) > 0 && (v[0] == '"' || v[0] == '\'') && (len(v) < 2 || v[len(v)-1] != v[0])) {
+ if !ok {
+ continue
+ }
+ if strings.HasSuffix(v, `\`) || (len(v) > 0 && (v[0] == '"' || v[0] == '\'') && (len(v) < 2 || v[len(v)-1] != v[0])) {
return claimUnitSettings{}, errors.New("unsupported multiline host unit setting")
}
if _, known := values[key]; !known {
diff --git a/internal/host/claimunit_test.go b/internal/host/claimunit_test.go
index 7eda7aff4..e86d71f47 100644
--- a/internal/host/claimunit_test.go
+++ b/internal/host/claimunit_test.go
@@ -108,6 +108,13 @@ func TestClaimControlUnitInstallers(t *testing.T) {
}
check(filepath.Join(stage, "etc/systemd/system", claimControlUnit))
f := newApplyFixture(t, "stable", "v1.0.0")
+ writeEnv(f.a.etcDir)
+ if err := os.Remove(filepath.Join(stage, "etc/carlos/host.env")); err != nil {
+ t.Fatal(err)
+ }
+ if err := bootstrap(stage, nil); err != nil {
+ t.Fatal(err)
+ }
if err := f.a.installUnits(t.Context(), stage, nil); err != nil {
t.Fatal(err)
}
@@ -255,8 +262,9 @@ func TestClaimBadUnitSettingsStillRepairSystemUnits(t *testing.T) {
probe := func(args ...string) (string, error) { return "inactive\n", nil }
ensureSystemUnits(etc, filepath.Join(t.TempDir(), "lock"), ctl, probe, t.Logf)
got, err := os.ReadFile(filepath.Join(units, claimControlUnit))
- if err != nil || !bytes.Equal(got, old) {
- t.Fatal("control unit changed on unknown settings")
+ want, _ := unitsFS.ReadFile("units/" + claimControlUnit)
+ if err != nil || !bytes.Equal(got, want) {
+ t.Fatal("control unit not restored to main on unknown settings")
}
for _, name := range systemUnitFiles {
if name != claimControlUnit {
diff --git a/internal/host/claimwakeassembly.go b/internal/host/claimwakeassembly.go
index 11332fd61..33e9fc158 100644
--- a/internal/host/claimwakeassembly.go
+++ b/internal/host/claimwakeassembly.go
@@ -131,10 +131,8 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
return w.Stop(ctx, o)
}
runtime.Prerequisites = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) error {
- if w.Roles != nil {
- if err := w.Roles.Preflight(call, ClaimLaunchInputs{}); err != nil {
- return err
- }
+ if err := w.Roles.Preflight(call, ClaimLaunchInputs{}); err != nil {
+ return err
}
if p.Probe || (o.Claim != nil && o.Claim.Op.Kind != hostclaim.KindServing) {
return &ClaimDependencyError{Contracts: []string{"moves C1 operation attempt journal and prepared-open adapter"}}
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index c5ea048d0..0fea54ddd 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -1173,6 +1173,10 @@ func (a *systemApplier) journal(j systemJournal) error {
// next boot — then daemon-reload. Never the staging root's data/: it holds
// a 0-byte registry.db.
func (a *systemApplier) installUnits(ctx context.Context, stagedRoot string, remove []string) error {
+ claimSettings, settingsErr := readClaimUnitSettings(filepath.Join(a.etcDir, "carlos", "host.env"))
+ if settingsErr != nil {
+ a.ops.logf("system apply: instance-control unit settings: %v", settingsErr)
+ }
unitDir := filepath.Join(a.etcDir, "systemd", "system")
if err := os.MkdirAll(unitDir, 0o755); err != nil {
return err
@@ -1195,6 +1199,16 @@ func (a *systemApplier) installUnits(ctx context.Context, stagedRoot string, rem
if e.IsDir() {
continue
}
+ if e.Name() == claimControlUnit {
+ b, err := os.ReadFile(filepath.Join(stagedRoot, "etc", "systemd", "system", e.Name()))
+ if err != nil {
+ return err
+ }
+ if err := writeFileDurable(filepath.Join(unitDir, e.Name()), claimUnitContent(e.Name(), b, claimSettings), 0644); err != nil {
+ return err
+ }
+ continue
+ }
if err := copyFileDurable(filepath.Join(stagedRoot, "etc", "systemd", "system", e.Name()), filepath.Join(unitDir, e.Name()), 0o644); err != nil {
return err
}
diff --git a/internal/host/systemunits.go b/internal/host/systemunits.go
index febfb54f5..1e912dd0c 100644
--- a/internal/host/systemunits.go
+++ b/internal/host/systemunits.go
@@ -94,9 +94,6 @@ func ensureSystemUnits(etcDir, lockPath string, systemctl, probe func(args ...st
isEnabled := parseIsEnabled(enabled)
missing := false
for _, u := range systemUnitFiles {
- if u == claimControlUnit && settingsErr != nil {
- continue
- }
if _, err := os.Stat(filepath.Join(unitDir, u)); err != nil {
missing = true
}
@@ -111,7 +108,7 @@ func ensureSystemUnits(etcDir, lockPath string, systemctl, probe func(args ...st
logf("adopt: instance-control unit: %v", err)
return
}
- controlChanged := settingsErr == nil && (err != nil || !bytes.Equal(installed, claimUnitContent(claimControlUnit, control, claimSettings)))
+ controlChanged := settingsErr == nil && (claimSettings.enabled || bytes.Contains(installed, []byte(claimControlEnvironment))) && (err != nil || !bytes.Equal(installed, claimUnitContent(claimControlUnit, control, claimSettings)))
if !missing && !controlChanged && *isActive && isEnabled != nil && *isEnabled && *ctlActive {
return
}
@@ -120,9 +117,6 @@ func ensureSystemUnits(etcDir, lockPath string, systemctl, probe func(args ...st
return
}
for _, u := range systemUnitFiles {
- if u == claimControlUnit && settingsErr != nil {
- continue
- }
b, err := unitsFS.ReadFile("units/" + u)
if err != nil {
logf("adopt: system units: %v", err)
diff --git a/internal/hostbox/coldhandover.go b/internal/hostbox/coldhandover.go
index bc0072b29..6b40dfaaf 100644
--- a/internal/hostbox/coldhandover.go
+++ b/internal/hostbox/coldhandover.go
@@ -511,7 +511,10 @@ func (j *LaunchJournal) HasInstallationEffects(ctx context.Context, host, incarn
if err = json.Unmarshal(body, &in); err != nil {
return false, err
}
- if in.Target.Host == host && in.Target.Incarnation == incarnation && in.Target.Stamp == stamp && (len(in.Reservations) > 0 || len(in.Files) > 0 || len(in.ProducerIDs) > 0) {
+ if !in.Completed.Valid() {
+ return false, ErrTypedInventory
+ }
+ if in.Target.Host == host && in.Target.Incarnation == incarnation && in.Target.Stamp == stamp && (in.Completed.Count > 0 || len(in.Reservations) > 0 || len(in.Files) > 0 || len(in.ProducerIDs) > 0) {
return true, nil
}
}
diff --git a/internal/hostbox/coldhandover_test.go b/internal/hostbox/coldhandover_test.go
index 8dd81c6d0..0ab0ac6e7 100644
--- a/internal/hostbox/coldhandover_test.go
+++ b/internal/hostbox/coldhandover_test.go
@@ -16,6 +16,7 @@ import (
)
func TestColdHandoverParksFilesRestoresOwnedRowsAndNeverReplaysOverStartedSuccessor(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
source, store, initial := workFixture(t)
ctx := t.Context()
admit := func(old hostclaim.Observed, generation string) hostclaim.Observed {
diff --git a/internal/hostbox/intended_recovery_test.go b/internal/hostbox/intended_recovery_test.go
index 2bbb975e5..9d6f636d1 100644
--- a/internal/hostbox/intended_recovery_test.go
+++ b/internal/hostbox/intended_recovery_test.go
@@ -34,6 +34,7 @@ type intendedFixture struct {
func newIntendedFixture(t *testing.T, edit func(*StopInventoryEntry)) intendedFixture {
t.Helper()
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
source, store, initial := workFixture(t)
ctx := t.Context()
admit := func(old hostclaim.Observed, generation string) hostclaim.Observed {
@@ -480,3 +481,24 @@ func TestColdHandoverEmptyDirectoryUsesCapturedIdentity(t *testing.T) {
t.Fatalf("nonempty directory accepted: %v", err)
}
}
+
+func TestClaimReviewHandoverRefusesCompletedSuccessorProducer(t *testing.T) {
+ f := newIntendedFixture(t, nil)
+ target := f.prior
+ target.Stamp = f.next.Stamp
+ target.Session = f.next.Session
+ j := f.handover.Launches
+ if err := j.Prelaunch(t.Context(), LaunchInventory{ControllerPID: 123, ProcessStart: "456", Target: target, Groups: []LaunchGroup{{RelativePath: "host", Device: 1, Inode: 1}}, NoState: true}); err != nil {
+ t.Fatal(err)
+ }
+ if err := j.RecordProducer(t.Context(), target, "delivery"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := j.CompleteProducer(t.Context(), target, "delivery", ProducerCompletion{}); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.handover.Run(t.Context(), f.prior, f.next); !errors.Is(err, ErrChanged) {
+ t.Fatalf("handover allowed completed successor: %v", err)
+ }
+ f.requireUnmoved(t)
+}
diff --git a/internal/hostbox/journalpath.go b/internal/hostbox/journalpath.go
new file mode 100644
index 000000000..68abfbf17
--- /dev/null
+++ b/internal/hostbox/journalpath.go
@@ -0,0 +1,22 @@
+package hostbox
+
+import (
+ "errors"
+ "os"
+)
+
+func checkJournalPath(path string) error {
+ for _, suffix := range []string{"", "-wal", "-shm"} {
+ info, err := os.Lstat(path + suffix)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return err
+ }
+ if !info.Mode().IsRegular() {
+ return ErrTypedInventory
+ }
+ }
+ return nil
+}
diff --git a/internal/hostbox/journalpath_test.go b/internal/hostbox/journalpath_test.go
new file mode 100644
index 000000000..d17c9ed8e
--- /dev/null
+++ b/internal/hostbox/journalpath_test.go
@@ -0,0 +1,42 @@
+package hostbox
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func TestClaimReviewJournalsRefusePlantedSymlinks(t *testing.T) {
+ for _, kind := range []string{"launch", "inventory"} {
+ for _, suffix := range []string{"", "-wal", "-shm"} {
+ t.Run(kind+suffix, func(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "journal.db")
+ victim := filepath.Join(dir, "victim")
+ if err := os.Symlink(victim, path+suffix); err != nil {
+ t.Fatal(err)
+ }
+ var err error
+ if kind == "launch" {
+ var j *LaunchJournal
+ j, err = OpenLaunchJournal(path)
+ if err == nil {
+ j.Close()
+ }
+ } else {
+ var j *StopInventoryDB
+ j, err = OpenStopInventoryDB(path)
+ if err == nil {
+ j.Close()
+ }
+ }
+ if err == nil {
+ t.Error("planted symlink accepted")
+ }
+ if _, err := os.Lstat(victim); !os.IsNotExist(err) {
+ t.Fatalf("SQLite touched symlink target: %v", err)
+ }
+ })
+ }
+ }
+}
diff --git a/internal/hostbox/launchjournal.go b/internal/hostbox/launchjournal.go
index a0d4532d9..ed1aaeeb7 100644
--- a/internal/hostbox/launchjournal.go
+++ b/internal/hostbox/launchjournal.go
@@ -42,6 +42,9 @@ func OpenLaunchJournal(path string) (*LaunchJournal, error) {
if !filepath.IsAbs(path) {
return nil, ErrUnsupported
}
+ if err := checkJournalPath(path); err != nil {
+ return nil, err
+ }
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, err
diff --git a/internal/hostbox/launchjournal_test.go b/internal/hostbox/launchjournal_test.go
index 3471a615b..501a78b6d 100644
--- a/internal/hostbox/launchjournal_test.go
+++ b/internal/hostbox/launchjournal_test.go
@@ -120,3 +120,30 @@ func TestLaunchJournalPrelaunchRecordIsRecoverableWithoutEffects(t *testing.T) {
t.Fatalf("launch files after pre-launch = %+v %v", got, err)
}
}
+
+func TestClaimReviewCompletedProducersAreInstallationEffects(t *testing.T) {
+ j, err := OpenLaunchJournal(filepath.Join(t.TempDir(), "launch.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer j.Close()
+ target := hostguard.FenceTarget{Host: "app.test", Incarnation: "inc", Stamp: 2, Session: 2, Box: "box", Enrollment: "enroll", BootID: "boot", ControllerID: "controller"}
+ if err := j.Prelaunch(t.Context(), LaunchInventory{ControllerPID: 123, ProcessStart: "456", Target: target, Groups: []LaunchGroup{{RelativePath: "host", Device: 1, Inode: 1}}, NoState: true}); err != nil {
+ t.Fatal(err)
+ }
+ if err := j.RecordProducer(t.Context(), target, "delivery"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := j.CompleteProducer(t.Context(), target, "delivery", ProducerCompletion{}); err != nil {
+ t.Fatal(err)
+ }
+ if effects, err := j.HasInstallationEffects(t.Context(), target.Host, target.Incarnation, target.Stamp); err != nil || !effects {
+ t.Fatalf("completed producer lost from handover check: effects=%v err=%v", effects, err)
+ }
+ if _, err := j.db.Exec(`UPDATE launches SET body=json_set(body,'$.Completed.Digest','broken')`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := j.HasInstallationEffects(t.Context(), target.Host, target.Incarnation, target.Stamp); err == nil {
+ t.Fatal("invalid completion summary accepted")
+ }
+}
diff --git a/internal/hostbox/stopinventory_db.go b/internal/hostbox/stopinventory_db.go
index 791dcf68a..7a8d759d4 100644
--- a/internal/hostbox/stopinventory_db.go
+++ b/internal/hostbox/stopinventory_db.go
@@ -15,6 +15,9 @@ func OpenStopInventoryDB(path string) (*StopInventoryDB, error) {
if !filepath.IsAbs(path) {
return nil, ErrTypedInventory
}
+ if err := checkJournalPath(path); err != nil {
+ return nil, err
+ }
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, err
diff --git a/internal/registry/claimhold.go b/internal/registry/claimhold.go
index 0d5f1390e..43eb8097f 100644
--- a/internal/registry/claimhold.go
+++ b/internal/registry/claimhold.go
@@ -9,7 +9,12 @@ func (r *Registry) SetClaimHold(ctx context.Context, host string, owner ClaimOwn
if !owner.Valid() || host == "" {
return ErrClaimOwnership
}
- result, err := r.sql.ExecContext(ctx, `UPDATE routes SET hold_op=? WHERE host=? AND claim_incarnation=? AND claim_stamp=? AND claim_owner=? AND claim_row_id=? AND hold_op=?`, next, host, owner.ClaimIncarnation, owner.ClaimStamp, owner.ClaimOwner, owner.ClaimRowID, expected)
+ tx, err := r.beginClaimTransaction(ctx, true)
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ result, err := tx.ExecContext(ctx, `UPDATE routes SET hold_op=? WHERE host=? AND claim_incarnation=? AND claim_stamp=? AND claim_owner=? AND claim_row_id=? AND hold_op=?`, next, host, owner.ClaimIncarnation, owner.ClaimStamp, owner.ClaimOwner, owner.ClaimRowID, expected)
if err != nil {
return err
}
@@ -20,5 +25,5 @@ func (r *Registry) SetClaimHold(ctx context.Context, host string, owner ClaimOwn
if n != 1 {
return ErrClaimOwnership
}
- return nil
+ return tx.Commit()
}
diff --git a/internal/registry/claimreview_test.go b/internal/registry/claimreview_test.go
new file mode 100644
index 000000000..51a7b241a
--- /dev/null
+++ b/internal/registry/claimreview_test.go
@@ -0,0 +1,167 @@
+package registry
+
+import (
+ "errors"
+ "fmt"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func TestClaimReviewOpenRetainsNormalDurability(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "")
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ var mode int
+ if err := r.sql.QueryRow("PRAGMA synchronous").Scan(&mode); err != nil {
+ t.Fatal(err)
+ }
+ if mode != 1 {
+ t.Fatalf("flag-off synchronous=%d, want NORMAL (1)", mode)
+ }
+}
+
+func TestClaimReviewFlagOffTimersKeepLegacyStatements(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "")
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ if _, err := r.sql.Exec("DROP TABLE routes"); err != nil {
+ t.Fatal(err)
+ }
+ timer := Timer{Host: "app.test", Name: "tick", Kind: "schedule", Spec: "every 1h0m0s", Path: "/tick", NextAt: 12, DueAt: 12}
+ if err := r.UpsertTimer(timer); err != nil {
+ t.Errorf("legacy upsert unexpectedly reads routes: %v", err)
+ }
+ timer.Registration = "imported"
+ if err := r.ReplaceTimers(timer.Host, []Timer{timer}); err != nil {
+ t.Fatalf("legacy replace unexpectedly reads routes: %v", err)
+ }
+ got, err := r.GetTimer(timer.Host, timer.Name)
+ if err != nil || *got != timer {
+ t.Fatalf("legacy import=%+v err=%v", got, err)
+ }
+}
+
+func TestClaimReviewTimerWritersWaitForOtherConnection(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ other, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer other.Close()
+ timer := Timer{Host: "app.test", Name: "tick", Kind: "schedule", Path: "/tick", Registration: "imported"}
+ for _, replace := range []bool{false, true} {
+ for i := 0; i < 100; i++ {
+ tx, err := other.sql.Begin()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := tx.Exec("UPDATE routes SET status=status"); err != nil {
+ t.Fatal(err)
+ }
+ done := make(chan error, 1)
+ go func() {
+ if replace {
+ done <- r.ReplaceTimers(timer.Host, []Timer{timer})
+ } else {
+ done <- r.UpsertTimer(timer)
+ }
+ }()
+ time.Sleep(time.Millisecond)
+ if err := tx.Commit(); err != nil {
+ t.Fatal(err)
+ }
+ if err := <-done; err != nil {
+ t.Fatalf("replace=%v writer failed instead of waiting: %v", replace, err)
+ }
+ }
+ }
+}
+
+func TestClaimReviewOwnershipMigrationDoesNotOwnHoldColumn(t *testing.T) {
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ if _, err := r.sql.Exec("ALTER TABLE routes DROP COLUMN hold_op"); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.migrateClaimOwnership(); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec("SELECT hold_op FROM routes"); err == nil {
+ t.Fatal("claim ownership migration redundantly added main's hold_op")
+ }
+}
+
+func TestClaimReviewDurabilityScopedToClaimWrites(t *testing.T) {
+ r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ rt := Route{Host: "app.test"}
+ if err := r.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ timer := Timer{Host: rt.Host, Name: "tick", Kind: "schedule"}
+ if err := r.UpsertTimer(timer); err != nil {
+ t.Fatal(err)
+ }
+ observed, err := r.Get(rt.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ observedTimer, err := r.GetTimer(rt.Host, timer.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, table := range []string{"routes", "timers", "claim_row_retirements"} {
+ for _, op := range []string{"UPDATE", "DELETE", "INSERT"} {
+ q := fmt.Sprintf(`CREATE TRIGGER require_full_%s_%s BEFORE %s ON %s BEGIN SELECT CASE WHEN (SELECT synchronous FROM pragma_synchronous) != 2 THEN RAISE(ABORT,'claim write without FULL durability') END; END`, table, op, op, table)
+ if _, err := r.sql.Exec(q); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ check := func(err error) {
+ t.Helper()
+ if err != nil {
+ t.Fatal(err)
+ }
+ var mode int
+ if err := r.sql.QueryRow("PRAGMA synchronous").Scan(&mode); err != nil || mode != 1 {
+ t.Fatalf("normal durability not restored: %d %v", mode, err)
+ }
+ }
+ owner := ClaimOwnership{"inc", 1, "owner", "route"}
+ check(r.AdoptClaimRoute(t.Context(), *observed, owner))
+ timerOwner := owner
+ timerOwner.ClaimRowID = observedTimer.Registration
+ check(r.AdoptClaimTimer(t.Context(), *observedTimer, timerOwner))
+ check(r.SetClaimHold(t.Context(), rt.Host, owner, "", "hold"))
+ if err := r.SetClaimHold(t.Context(), rt.Host, owner, "wrong", ""); err != ErrClaimOwnership {
+ t.Fatal(err)
+ }
+ check(nil)
+ targets := []ClaimRow{{Kind: "route", Host: rt.Host, Ownership: owner}, {Kind: "timer", Host: rt.Host, Name: timer.Name, Registration: observedTimer.Registration, Ownership: timerOwner}}
+ failed := errors.New("archive failed")
+ if err := r.RetireClaimRows(t.Context(), "retire", targets, func([]ClaimRowArchive) error { return failed }); !errors.Is(err, failed) {
+ t.Fatal(err)
+ }
+ check(nil)
+ check(r.RetireClaimRows(t.Context(), "retire", targets, func([]ClaimRowArchive) error { return nil }))
+}
diff --git a/internal/registry/claimrows.go b/internal/registry/claimrows.go
index 5d5f0342b..f581b8901 100644
--- a/internal/registry/claimrows.go
+++ b/internal/registry/claimrows.go
@@ -36,9 +36,6 @@ type ClaimRowArchive struct {
}
func (r *Registry) migrateClaimOwnership() error {
- if _, err := r.sql.Exec("ALTER TABLE routes ADD COLUMN hold_op TEXT NOT NULL DEFAULT ''"); err != nil && !strings.Contains(err.Error(), "duplicate column") {
- return err
- }
for _, table := range []string{"routes", "timers"} {
for _, column := range []string{"claim_incarnation TEXT NOT NULL DEFAULT ''", "claim_stamp INTEGER NOT NULL DEFAULT 0", "claim_owner TEXT NOT NULL DEFAULT ''", "claim_row_id TEXT NOT NULL DEFAULT ''"} {
if _, err := r.sql.Exec("ALTER TABLE " + table + " ADD COLUMN " + column); err != nil && !strings.Contains(err.Error(), "duplicate column") {
@@ -59,7 +56,7 @@ func (r *Registry) AdoptClaimRoute(ctx context.Context, observed Route, owner Cl
if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) {
return ErrClaimOwnership
}
- tx, err := r.sql.BeginTx(ctx, nil)
+ tx, err := r.beginClaimTransaction(ctx, true)
if err != nil {
return err
}
@@ -94,7 +91,7 @@ func (r *Registry) RetireClaimRows(ctx context.Context, journal string, targets
if journal == "" || len(targets) == 0 || archive == nil {
return ErrClaimOwnership
}
- tx, err := r.sql.BeginTx(ctx, nil)
+ tx, err := r.beginClaimTransaction(ctx, true)
if err != nil {
return err
}
@@ -195,7 +192,7 @@ func (r *Registry) AdoptClaimTimer(ctx context.Context, observed Timer, owner Cl
if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) || observed.Registration == "" || owner.ClaimRowID != observed.Registration {
return ErrClaimOwnership
}
- tx, err := r.sql.BeginTx(ctx, nil)
+ tx, err := r.beginClaimTransaction(ctx, true)
if err != nil {
return err
}
diff --git a/internal/registry/claimrows_test.go b/internal/registry/claimrows_test.go
index da67232a1..6ad9c2c0c 100644
--- a/internal/registry/claimrows_test.go
+++ b/internal/registry/claimrows_test.go
@@ -8,6 +8,7 @@ import (
)
func TestClaimRowsArchiveBeforeDeleteAndReplayPreservesSuccessor(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
ctx := context.Background()
path := filepath.Join(t.TempDir(), "registry.db")
r, err := Open(path)
@@ -81,6 +82,7 @@ func TestClaimRowsArchiveBeforeDeleteAndReplayPreservesSuccessor(t *testing.T) {
}
}
func TestClaimRowsRefusesChangedRegistrationAtomically(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
if err != nil {
t.Fatal(err)
diff --git a/internal/registry/claimtransaction.go b/internal/registry/claimtransaction.go
new file mode 100644
index 000000000..de7f07b31
--- /dev/null
+++ b/internal/registry/claimtransaction.go
@@ -0,0 +1,74 @@
+package registry
+
+import (
+ "context"
+ "database/sql"
+ "database/sql/driver"
+ "errors"
+)
+
+type claimTransaction struct {
+ conn *sql.Conn
+ ctx context.Context
+ full bool
+ done bool
+}
+
+func (r *Registry) beginClaimTransaction(ctx context.Context, full bool) (*claimTransaction, error) {
+ conn, err := r.sql.Conn(ctx)
+ if err != nil {
+ return nil, err
+ }
+ tx := &claimTransaction{conn: conn, ctx: ctx, full: full}
+ if full {
+ if _, err = conn.ExecContext(ctx, "PRAGMA synchronous=FULL"); err != nil {
+ return nil, errors.Join(err, tx.close())
+ }
+ }
+ if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
+ return nil, errors.Join(err, tx.close())
+ }
+ return tx, nil
+}
+func (t *claimTransaction) ExecContext(ctx context.Context, q string, args ...any) (sql.Result, error) {
+ return t.conn.ExecContext(ctx, q, args...)
+}
+func (t *claimTransaction) QueryRowContext(ctx context.Context, q string, args ...any) *sql.Row {
+ return t.conn.QueryRowContext(ctx, q, args...)
+}
+func (t *claimTransaction) Exec(q string, args ...any) (sql.Result, error) {
+ return t.ExecContext(t.ctx, q, args...)
+}
+func (t *claimTransaction) QueryRow(q string, args ...any) *sql.Row {
+ return t.QueryRowContext(t.ctx, q, args...)
+}
+func (t *claimTransaction) close() error {
+ t.done = true
+ var err error
+ if t.full {
+ _, err = t.conn.ExecContext(context.Background(), "PRAGMA synchronous=NORMAL")
+ if err != nil {
+ _ = t.conn.Raw(func(any) error { return driver.ErrBadConn })
+ }
+ }
+ return errors.Join(err, t.conn.Close())
+}
+func (t *claimTransaction) Commit() error {
+ if t.done {
+ return sql.ErrTxDone
+ }
+ if _, err := t.conn.ExecContext(t.ctx, "COMMIT"); err != nil {
+ return errors.Join(err, t.Rollback())
+ }
+ return t.close()
+}
+func (t *claimTransaction) Rollback() error {
+ if t.done {
+ return nil
+ }
+ _, err := t.conn.ExecContext(context.Background(), "ROLLBACK")
+ if err != nil {
+ _ = t.conn.Raw(func(any) error { return driver.ErrBadConn })
+ }
+ return errors.Join(err, t.close())
+}
diff --git a/internal/registry/coldhandover.go b/internal/registry/coldhandover.go
index c6499e6ee..6e832076a 100644
--- a/internal/registry/coldhandover.go
+++ b/internal/registry/coldhandover.go
@@ -35,7 +35,7 @@ func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string,
if _, err := r.sql.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS cold_handovers(id TEXT PRIMARY KEY,body BLOB NOT NULL)`); err != nil {
return err
}
- tx, err := r.sql.BeginTx(ctx, nil)
+ tx, err := r.beginClaimTransaction(ctx, true)
if err != nil {
return err
}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 1e05b8394..79a8353ee 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -445,7 +445,7 @@ func Open(path string) (*Registry, error) {
for _, p := range []string{
"PRAGMA busy_timeout=5000",
"PRAGMA journal_mode=WAL",
- "PRAGMA synchronous=FULL",
+ "PRAGMA synchronous=NORMAL",
} {
if _, err := d.Exec(p); err != nil {
return nil, err
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index e385ac9ac..57d7d2fac 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -8,11 +8,13 @@
package registry
import (
+ "context"
"crypto/rand"
"database/sql"
"encoding/hex"
"errors"
"fmt"
+ "github.com/carlosframework/platform/internal/hostwrite"
"os"
"strings"
)
@@ -49,7 +51,10 @@ 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 {
- tx, err := r.sql.Begin()
+ if !hostwrite.Enabled(os.Getenv) {
+ return r.upsertLegacyTimer(t)
+ }
+ tx, err := r.beginClaimTransaction(context.Background(), false)
if err != nil {
return err
}
@@ -77,7 +82,9 @@ func (r *Registry) UpsertTimer(t Timer) error {
}
return tx.Commit()
}
-func upsertTimer(tx *sql.Tx, t Timer) error {
+func upsertTimer(tx interface {
+ Exec(string, ...any) (sql.Result, error)
+}, t Timer) error {
const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
res, err := tx.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(host, name) DO UPDATE SET
@@ -111,12 +118,15 @@ 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 !hostwrite.Enabled(os.Getenv) {
+ return r.replaceLegacyTimers(host, rows)
+ }
for _, t := range rows {
if t.Host != host {
return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host)
}
}
- tx, err := r.sql.Begin()
+ tx, err := r.beginClaimTransaction(context.Background(), false)
if err != nil {
return err
}
@@ -336,3 +346,52 @@ func (r *Registry) EarliestTimers() (map[string]int64, error) {
}
return out, rows.Err()
}
+
+const legacyTimerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration`
+
+func (r *Registry) upsertLegacyTimer(t Timer) error {
+ const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
+ res, err := r.sql.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16))))
+ON CONFLICT(host, name) DO UPDATE SET
+ spec=excluded.spec, path=excluded.path,
+ next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
+ due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
+ attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
+ registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END
+WHERE timers.kind=excluded.kind`,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return ErrTimerKind
+ }
+ return nil
+}
+
+func (r *Registry) replaceLegacyTimers(host string, rows []Timer) error {
+ for _, t := range rows {
+ if t.Host != host {
+ return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host)
+ }
+ }
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
+ return err
+ }
+ for _, t := range rows {
+ if _, err := tx.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration); err != nil {
+ return fmt.Errorf("registry: insert timer %q: %w", t.Name, err)
+ }
+ }
+ return tx.Commit()
+}