amadan

carlos / platform Public

Serialize scheduled token preparation with maintenance fence

Paul Campbell pushed by paul@keymail.dev 2abaf89e8556067c3e459cfeaedfa3a5dc1c8f7a parent fe90ceb
6 files changed, +225 −8
  • internal/activator/scheduled_token.go +17 −0
  • internal/activator/scheduled_token_test.go +90 −0
  • internal/host/schedule_token_fence_test.go +102 −0
  • internal/host/schedule_wake_test.go +2 −2
  • internal/host/schedule_wiring.go +11 −3
  • internal/host/schedule_wiring_test.go +3 −3
diff --git a/internal/activator/scheduled_token.go b/internal/activator/scheduled_token.go
new file mode 100644
index 000000000..0074b6320
--- /dev/null
+++ b/internal/activator/scheduled_token.go
@@ -0,0 +1,17 @@
+package activator
+
+// PrepareScheduledToken serializes token preparation with the local maintenance
+// fence. A preparation already in progress finishes before the fence is installed;
+// a fenced host never enters prepare. The callback must not reenter Service.
+// This does not authorize a wake or establish durable maintenance authority.
+func (a *Service) PrepareScheduledToken(host string, prepare func() (string, error)) (string, error) {
+ if a == nil || prepare == nil {
+ return "", ErrFenced
+ }
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if a.fences[host] != 0 {
+ return "", ErrFenced
+ }
+ return prepare()
+}
diff --git a/internal/activator/scheduled_token_test.go b/internal/activator/scheduled_token_test.go
new file mode 100644
index 000000000..ee2841e43
--- /dev/null
+++ b/internal/activator/scheduled_token_test.go
@@ -0,0 +1,90 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+)
+
+func TestScheduledTokenFenceRefusesPreparation(t *testing.T) {
+ a := &Service{}
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if _, err := a.FenceAndStop(ctx, "token.example.com", 1); !errors.Is(err, ErrStopUnproven) {
+ t.Fatalf("missing entry: %v", err)
+ }
+ called := false
+ tok, err := a.PrepareScheduledToken("token.example.com", func() (string, error) {
+ called = true
+ return "secret", nil
+ })
+ if called || tok != "" || !errors.Is(err, ErrFenced) {
+ t.Fatalf("fenced preparation: called=%v token-present=%v err=%v", called, tok != "", err)
+ }
+ want := errors.New("prepare failed")
+ _, err = a.PrepareScheduledToken("sibling.example.com", func() (string, error) { return "", want })
+ if !errors.Is(err, want) {
+ t.Fatalf("unfenced sibling: %v", err)
+ }
+}
+
+func TestScheduledTokenPreparationOrdersFence(t *testing.T) {
+ const host = "token.example.com"
+ a := &Service{stopHistory: map[string]*stopObservation{host: {
+ mainStop: true, mainExited: true, sidecarStop: true, sidecarExited: true, replicaClosed: true,
+ }}}
+ entered, release := make(chan struct{}), make(chan struct{})
+ done := make(chan error, 1)
+ go func() {
+ _, err := a.PrepareScheduledToken(host, func() (string, error) {
+ close(entered)
+ <-release
+ return "token", nil
+ })
+ done <- err
+ }()
+ select {
+ case <-entered:
+ case <-time.After(time.Second):
+ t.Fatal("preparation did not start")
+ }
+ // Even with prior checked-stop history, a deadline cannot establish a
+ // fence while previously admitted token I/O is still running.
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+ proof, err := a.FenceAndStop(ctx, host, 1)
+ cancel()
+ close(release)
+ if proof != nil || !errors.Is(err, ErrStopUnproven) {
+ t.Fatalf("fence crossed active preparation: proof=%v err=%v", proof != nil, err)
+ }
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("preparation did not finish")
+ }
+ ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ proof, err = a.FenceAndStop(ctx, host, 1)
+ if err != nil || !proof.Matches(host, 1) {
+ t.Fatalf("fence after preparation: %v", err)
+ }
+ _, err = a.PrepareScheduledToken(host, func() (string, error) {
+ t.Error("post-fence callback ran")
+ return "", nil
+ })
+ if !errors.Is(err, ErrFenced) {
+ t.Fatal(err)
+ }
+}
+
+func TestScheduledTokenInvalidCallback(t *testing.T) {
+ for _, a := range []*Service{nil, {}} {
+ if _, err := a.PrepareScheduledToken("token.example.com", nil); !errors.Is(err, ErrFenced) {
+ t.Fatalf("nil callback: %v", err)
+ }
+ }
+}
diff --git a/internal/host/schedule_token_fence_test.go b/internal/host/schedule_token_fence_test.go
new file mode 100644
index 000000000..a902c1bfe
--- /dev/null
+++ b/internal/host/schedule_token_fence_test.go
@@ -0,0 +1,102 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+)
+
+func TestScheduleTokenFencePreservesEnvironment(t *testing.T) {
+ for _, content := range []string{"", "APP_SETTING=retained\n", "APP_SETTING=retained\nCARLOS_ADMIN_TOKEN=" + strings.Repeat("a", 64) + "\n"} {
+ name := "missing-file"
+ if content != "" {
+ name = "existing-without-token"
+ }
+ if strings.Contains(content, "CARLOS_ADMIN_TOKEN") {
+ name = "existing-token"
+ }
+ t.Run(name, func(t *testing.T) {
+ dd := t.TempDir()
+ const host = "fenced.example.com"
+ path := filepath.Join(dd, host+".env")
+ if content != "" {
+ if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(path, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ svc := &activator.Service{}
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if _, err := svc.FenceAndStop(ctx, host, 1); !errors.Is(err, activator.ErrStopUnproven) {
+ t.Fatal(err)
+ }
+ // Exercise the actual runner wiring, including the nil-store arm.
+ sd := newScheduleDeps(nil, nil, svc, nil, dd)
+ if token, err := sd.runner.Token(host); token != "" || !errors.Is(err, activator.ErrFenced) {
+ t.Fatalf("fenced token: present=%v err=%v", token != "", err)
+ }
+ raw, err := os.ReadFile(path)
+ if content == "" {
+ if !os.IsNotExist(err) {
+ t.Fatalf("fence created env: %v", err)
+ }
+ return
+ }
+ if err != nil || !bytes.Equal(raw, []byte(content)) {
+ t.Fatalf("fence changed bytes: %v", err)
+ }
+ fi, err := os.Stat(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if fi.Mode().Perm() != 0o644 {
+ t.Fatalf("fence changed mode: %o", fi.Mode().Perm())
+ }
+ })
+ }
+}
+
+func TestScheduleTokenUnfencedManagedPreparation(t *testing.T) {
+ dd := t.TempDir()
+ const host = "ordinary.example.com"
+ path := filepath.Join(dd, host+".env")
+ if err := os.WriteFile(path, []byte("APP_SETTING=retained\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ sd := newScheduleDeps(nil, nil, &activator.Service{}, nil, dd)
+ first, err := sd.runner.Token(host)
+ if err != nil || len(first) != 64 {
+ t.Fatalf("mint: present=%v err=%v", first != "", err)
+ }
+ if err := os.Chmod(path, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ second, err := sd.runner.Token(host)
+ if err != nil || first != second {
+ t.Fatalf("retained token: %v", err)
+ }
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.HasPrefix(raw, []byte("APP_SETTING=retained\n")) || bytes.Count(raw, []byte("CARLOS_ADMIN_TOKEN=")) != 1 {
+ t.Fatal("ordinary token preparation lost environment or duplicated token")
+ }
+ fi, err := os.Stat(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if fi.Mode().Perm() != 0o600 {
+ t.Fatalf("ordinary mode: %o", fi.Mode().Perm())
+ }
+}
diff --git a/internal/host/schedule_wake_test.go b/internal/host/schedule_wake_test.go
index ce1ee966d..691c80e62 100644
--- a/internal/host/schedule_wake_test.go
+++ b/internal/host/schedule_wake_test.go
@@ -238,7 +238,7 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
now := time.Now().UTC()
occurrence := now.Unix() - 5
r := &schedule.Runner{
- Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd),
+ Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc),
Now: func() time.Time { return now }, Timeout: 30 * time.Second,
Box: "box1", Build: "test",
}
@@ -257,7 +257,7 @@ func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
case <-time.After(30 * time.Second):
t.Fatal("the handler never ran — the tick did not wake the sleeping instance")
}
- tok, err := scheduleToken(dd)(host)
+ tok, err := scheduleToken(dd, svc)(host)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go
index 266f3a434..6774ca4c1 100644
--- a/internal/host/schedule_wiring.go
+++ b/internal/host/schedule_wiring.go
@@ -84,7 +84,7 @@ func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator
runner: &schedule.Runner{
Reg: reg,
Deliver: &activatorDeliverer{svc: svc},
- Token: scheduleToken(dd),
+ Token: scheduleToken(dd, svc),
Event: event,
Box: box,
Build: version.Build,
@@ -157,9 +157,9 @@ func armSchedule(reg *registry.Registry, store release.Store, storeErr error, sv
// The error is redacted: it reaches the row's last_status and from there the
// bucket status object and `carlos schedule ls`, and a member must not learn
// the box's filesystem layout from it.
-func scheduleToken(dd string) func(host string) (string, error) {
+func scheduleToken(dd string, svc *activator.Service) func(host string) (string, error) {
read := schedule.TokenFromEnvFile(dd)
- return func(host string) (string, error) {
+ prepare := func(host string) (string, error) {
if err := activator.EnsureAdminToken(filepath.Join(dd, host+".env")); err != nil {
return "", errors.New(activator.RedactPaths(err.Error()))
}
@@ -169,6 +169,14 @@ func scheduleToken(dd string) func(host string) (string, error) {
}
return tok, nil
}
+ return func(host string) (string, error) {
+ if svc == nil {
+ return prepare(host)
+ }
+ return svc.PrepareScheduledToken(host, func() (string, error) {
+ return prepare(host)
+ })
+ }
}
// convergePassTimeout bounds ONE converge pass. The agent's three call sites
diff --git a/internal/host/schedule_wiring_test.go b/internal/host/schedule_wiring_test.go
index 8e5940f75..ea89a293e 100644
--- a/internal/host/schedule_wiring_test.go
+++ b/internal/host/schedule_wiring_test.go
@@ -121,7 +121,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
if _, err := os.Stat(envPath); !os.IsNotExist(err) {
t.Fatalf("fixture: env file already exists (%v)", err)
}
- tok, err := scheduleToken(dd)(host)
+ tok, err := scheduleToken(dd, nil)(host)
if err != nil {
t.Fatalf("Token on a host with no env file: %v", err)
}
@@ -135,7 +135,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
if err := activator.EnsureAdminToken(envPath); err != nil {
t.Fatal(err)
}
- again, err := scheduleToken(dd)(host)
+ again, err := scheduleToken(dd, nil)(host)
if err != nil || again != tok {
t.Fatalf("second read = %q (%v), want the first token %q", again, err, tok)
}
@@ -150,7 +150,7 @@ func TestScheduleTokenMintsForASleepingInstance(t *testing.T) {
// filesystem layout.
func TestScheduleTokenErrorCarriesNoAbsolutePath(t *testing.T) {
dd := filepath.Join(t.TempDir(), "missing-data-dir")
- _, err := scheduleToken(dd)("x.bab.oncarlos.com")
+ _, err := scheduleToken(dd, nil)("x.bab.oncarlos.com")
if err == nil {
t.Fatal("Token succeeded with no data dir")
}