carlos / platform Public

ops litestream: the cap is sized from the config on disk, under the lock (#284)

Summary

Astra's fourth pass on slice A: one P2. Of two overlapping runs, the
one that took the drop-in lock last sized the cap from the database set
it had counted before queueing, and could leave a config naming 200
databases under a cap fitted to 32. The count is now read from the
config on disk inside the lock.
Census overlay regenerated: eight rows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 week ago 2ec2e90d4e3866479b883af23976caba3485ecfe parent 01e1269

5 files changed, +111 −32

internal/ops/litestream_memory.go +30 −11
internal/ops/ops.go +3 −3
internal/ops/ops_test.go +69 −11
internal/provision/plan14_root_reads_test.go +2 −2
internal/provision/testdata/plan14-root-reads-ls284.tsv +7 −5

CI cancelled — superseded by a newer push — run details

CI log

=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.011s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... -count=1 -timeout 30m

amadan: job cancelled — a newer push superseded it

diff --git a/internal/ops/litestream_memory.go b/internal/ops/litestream_memory.go

index 6a8e9fd78..ad680b14b 100644

--- a/internal/ops/litestream_memory.go

+++ b/internal/ops/litestream_memory.go

@@ -94,43 +94,62 @@ func LitestreamMemoryDropIn(n int) string {

// The whole of it runs under a lock on the drop-in's directory. The verb is

// run by hand and may be run by a timer, and two runs interleaved could have

// one remove the marker the other had just left for a reload that failed.

-func writeMemoryDropIn(path string, n int, reload func() error) (changed bool, err error) {

- want := LitestreamMemoryDropIn(n)

+//

+// The number of databases is asked for inside the lock, through count, and

+// the verb answers from the config on disk at that moment, not from the set

+// it computed before it queued. Of two overlapping runs the one that takes

+// the lock last then sizes the cap for the config that is actually there,

+// whichever of them wrote it; a run that sized from its own earlier count

+// could leave a config naming 200 databases under a cap fitted to 32.

+func writeMemoryDropIn(path string, count func() (int, error), reload func() error) (n int, changed bool, err error) {

pending := path + ".reload-pending"

if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {

- return false, err

+ return 0, false, err

}

dir, err := os.Open(filepath.Dir(path))

if err != nil {

- return false, err

+ return 0, false, err

}

defer dir.Close()

if err := lockDir(dir); err != nil {

- return false, err

+ return 0, false, err

}

defer unlockDir(dir)

+ if n, err = count(); err != nil {

+ return 0, false, err

+ }

+ want := LitestreamMemoryDropIn(n)

prior, readErr := os.ReadFile(path)

if readErr != nil && !os.IsNotExist(readErr) {

- return false, readErr

+ return n, false, readErr

}

current := readErr == nil && string(prior) == want

if _, err := os.Lstat(pending); current && os.IsNotExist(err) {

- return false, nil

+ return n, false, nil

}

if err := safefile.WriteFile(pending, nil, 0o644); err != nil {

- return false, err

+ return n, false, err

}

if !current {

if err := safefile.WriteFile(path, []byte(want), 0o644); err != nil {

- return false, err

+ return n, false, err

}

if err := reload(); err != nil {

- return false, fmt.Errorf("systemd did not reload: %w", err)

+ return n, false, fmt.Errorf("systemd did not reload: %w", err)

}

// systemd has the cap. A marker that will not go only costs the next run

// one reload it did not need.

_ = os.Remove(pending)

- return true, nil

+ return n, true, nil

+// configDBCount counts the databases a generated config on disk names.

+func configDBCount(cfgPath string) (int, error) {

+ b, err := os.ReadFile(cfgPath)

+ if err != nil {

+ return 0, err

+ }

+ return len(ParseLitestreamDBs(string(b))), nil

}

diff --git a/internal/ops/ops.go b/internal/ops/ops.go

index f2d04b349..6cd8ed81c 100644

--- a/internal/ops/ops.go

+++ b/internal/ops/ops.go

@@ -155,7 +155,7 @@ should no longer be replicated.`,

}

fmt.Fprintf(os.Stderr, "litestream: %d databases -> s3://%s/%s/ (%s)\n", len(dbs), *bucket, *prefix, *out)

- // The cap follows the count this run just wrote (see litestream_memory.go).

+ // The cap follows the config now on disk (see litestream_memory.go).

// A run writing somewhere other than the live config is a preview or a

// test, and leaves systemd alone unless told where to write.

dropIn := *memDropIn

@@ -163,7 +163,7 @@ should no longer be replicated.`,

dropIn = defaultMemoryDropIn

}

if dropIn != "" && dropIn != "none" {

- capped, err := writeMemoryDropIn(dropIn, len(dbs), systemdDaemonReload)

+ n, capped, err := writeMemoryDropIn(dropIn, func() (int, error) { return configDBCount(*out) }, systemdDaemonReload)

if err != nil {

// Loud, not fatal: the config above is written and correct, and

// refusing the restart it needs would trade a cap that may be too

@@ -173,7 +173,7 @@ should no longer be replicated.`,

fmt.Fprintf(os.Stderr, "litestream: WARNING: could not set the memory cap in %s: %v\n The replicator keeps its current cap. The next run will try again.\n", dropIn, err)

}

if capped {

- fmt.Fprintf(os.Stderr, "litestream: memory cap %dM for %d databases (%s)\n", LitestreamMemoryMax(len(dbs))>>20, len(dbs), dropIn)

+ fmt.Fprintf(os.Stderr, "litestream: memory cap %dM for %d databases (%s)\n", LitestreamMemoryMax(n)>>20, n, dropIn)

wrote = true

}

diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go

index 3262997e1..4353dbd12 100644

--- a/internal/ops/ops_test.go

+++ b/internal/ops/ops_test.go

@@ -924,7 +924,7 @@ func TestWriteMemoryDropInReloadsOnlyOnChange(t *testing.T) {

reloads := 0

reload := func() error { reloads++; return nil }

- changed, err := writeMemoryDropIn(path, 100, reload)

+ _, changed, err := writeMemoryDropIn(path, dbs(100), reload)

if err != nil || !changed || reloads != 1 {

t.Fatalf("first write: changed=%v err=%v reloads=%d, want true, nil, 1", changed, err, reloads)

}

@@ -945,11 +945,11 @@ func TestWriteMemoryDropInReloadsOnlyOnChange(t *testing.T) {

}

// The same count changes no byte, so it reloads nothing.

- if changed, err := writeMemoryDropIn(path, 100, reload); err != nil || changed || reloads != 1 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(100), reload); err != nil || changed || reloads != 1 {

t.Fatalf("same count: changed=%v err=%v reloads=%d, want false, nil, 1", changed, err, reloads)

}

- if changed, err := writeMemoryDropIn(path, 101, reload); err != nil || !changed || reloads != 2 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(101), reload); err != nil || !changed || reloads != 2 {

t.Fatalf("one more database: changed=%v err=%v reloads=%d, want true, nil, 2", changed, err, reloads)

}

@@ -966,7 +966,7 @@ func TestWriteMemoryDropInRetriesAfterAFailedReload(t *testing.T) {

working := func() error { reloads++; return nil }

want := "MemoryMax=" + strconv.FormatInt(LitestreamMemoryMax(200)>>20, 10) + "M\n"

- if changed, err := writeMemoryDropIn(path, 200, broken); changed || err == nil {

+ if _, changed, err := writeMemoryDropIn(path, dbs(200), broken); changed || err == nil {

t.Fatalf("changed=%v err=%v, want false and the reload error", changed, err)

}

if body, _ := os.ReadFile(path); !strings.Contains(string(body), want) {

@@ -976,16 +976,16 @@ func TestWriteMemoryDropInRetriesAfterAFailedReload(t *testing.T) {

t.Fatalf("a failed reload left no marker: %v", err)

}

// The bytes are already right, and it still reloads, twice if it has to.

- if changed, err := writeMemoryDropIn(path, 200, broken); changed || err == nil || reloads != 2 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(200), broken); changed || err == nil || reloads != 2 {

t.Fatalf("second failure: changed=%v err=%v reloads=%d, want false, an error, 2", changed, err, reloads)

}

- if changed, err := writeMemoryDropIn(path, 200, working); !changed || err != nil || reloads != 3 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(200), working); !changed || err != nil || reloads != 3 {

t.Fatalf("retry: changed=%v err=%v reloads=%d, want true, nil, 3", changed, err, reloads)

}

if _, err := os.Stat(pending); !os.IsNotExist(err) {

t.Fatalf("the marker outlived a successful reload (stat err %v)", err)

}

- if changed, err := writeMemoryDropIn(path, 200, working); changed || err != nil || reloads != 3 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(200), working); changed || err != nil || reloads != 3 {

t.Fatalf("settled: changed=%v err=%v reloads=%d, want false, nil, 3", changed, err, reloads)

}

@@ -994,7 +994,7 @@ func TestWriteMemoryDropInRetriesAfterAFailedReload(t *testing.T) {

if err := os.WriteFile(pending, nil, 0o644); err != nil {

t.Fatal(err)

}

- if changed, err := writeMemoryDropIn(path, 200, working); !changed || err != nil || reloads != 4 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(200), working); !changed || err != nil || reloads != 4 {

t.Fatalf("stray marker: changed=%v err=%v reloads=%d, want true, nil, 4", changed, err, reloads)

}

@@ -1008,7 +1008,7 @@ func TestWriteMemoryDropInRunsOneAtATime(t *testing.T) {

second := make(chan error, 1)

firstReload := func() error {

go func() {

- _, err := writeMemoryDropIn(path, 300, func() error { return errors.New("no systemd here") })

+ _, _, err := writeMemoryDropIn(path, dbs(300), func() error { return errors.New("no systemd here") })

second <- err

}()

select {

@@ -1018,7 +1018,7 @@ func TestWriteMemoryDropInRunsOneAtATime(t *testing.T) {

}

return nil

}

- if changed, err := writeMemoryDropIn(path, 100, firstReload); !changed || err != nil {

+ if _, changed, err := writeMemoryDropIn(path, dbs(100), firstReload); !changed || err != nil {

t.Fatalf("first run: changed=%v err=%v, want true, nil", changed, err)

}

select {

@@ -1034,7 +1034,65 @@ func TestWriteMemoryDropInRunsOneAtATime(t *testing.T) {

t.Fatalf("the second run's marker is gone: %v", err)

}

reloads := 0

- if changed, err := writeMemoryDropIn(path, 300, func() error { reloads++; return nil }); !changed || err != nil || reloads != 1 {

+ if _, changed, err := writeMemoryDropIn(path, dbs(300), func() error { reloads++; return nil }); !changed || err != nil || reloads != 1 {

t.Fatalf("the run after: changed=%v err=%v reloads=%d, want true, nil, 1", changed, err, reloads)

}

+// dbs is a config that names n databases, for writeMemoryDropIn's count.

+func dbs(n int) func() (int, error) { return func() (int, error) { return n, nil } }

+// The cap is sized from the config that is on disk when the run holds the

+// lock, not from what the run counted before it queued. Run A counted 32

+// databases, then waited behind run B, which wrote a config of 200 and its

+// cap. A must leave the cap for 200, not put back the one for 32.

+func TestWriteMemoryDropInSizesFromTheConfigUnderTheLock(t *testing.T) {

+ dir := t.TempDir()

+ path := filepath.Join(dir, "d", "50-memory.conf")

+ cfg := filepath.Join(dir, "litestream.yml")

+ write := func(n int) {

+ paths := make([]string, n)

+ for i := range paths {

+ paths[i] = "/data/db" + strconv.Itoa(i) + ".db"

+ }

+ if err := os.WriteFile(cfg, []byte(LitestreamConfig(paths, "bucket", "prefix", "eu-west-1")), 0o644); err != nil {

+ t.Fatal(err)

+ }

+ count := func() (int, error) { return configDBCount(cfg) }

+ write(32) // A's config, written before it queues for the lock

+ aDone := make(chan int, 1)

+ bReload := func() error {

+ // B holds the lock. A arrives, having counted 32, and waits.

+ go func() {

+ n, _, err := writeMemoryDropIn(path, count, func() error { return nil })

+ if err != nil {

+ t.Error(err)

+ }

+ aDone <- n

+ }()

+ time.Sleep(200 * time.Millisecond)

+ return nil

+ }

+ write(200) // B's config lands, then B takes the lock and caps it

+ if n, changed, err := writeMemoryDropIn(path, count, bReload); n != 200 || !changed || err != nil {

+ t.Fatalf("B: n=%d changed=%v err=%v, want 200, true, nil", n, changed, err)

+ }

+ select {

+ case n := <-aDone:

+ if n != 200 {

+ t.Fatalf("A sized the cap for %d databases, want the 200 the config names", n)

+ }

+ case <-time.After(10 * time.Second):

+ t.Fatal("A never ran")

+ }

+ body, _ := os.ReadFile(path)

+ if want := "MemoryMax=" + strconv.FormatInt(LitestreamMemoryMax(200)>>20, 10) + "M\n"; !strings.Contains(string(body), want) {

+ t.Fatalf("the cap left on disk is not the one for 200 databases:\n%s", body)

+ }

+ if _, _, err := writeMemoryDropIn(path, func() (int, error) { return 0, errors.New("config unreadable") }, func() error { return nil }); err == nil {

+ t.Fatal("a count that fails must fail the cap, not size it for zero databases")

+ }

diff --git a/internal/provision/plan14_root_reads_test.go b/internal/provision/plan14_root_reads_test.go

index ba614f33e..a567bad03 100644

--- a/internal/provision/plan14_root_reads_test.go

+++ b/internal/provision/plan14_root_reads_test.go

@@ -665,9 +665,9 @@ func TestPlan14RootReadCensus(t *testing.T) {

t.Fatal(err)

}

// Census ruling ls284 (#284): the litestream memory drop-in, the unit's

- // restart backoff and the agent's restart-counter probe and its saved state add seven frozen

+ // restart backoff and the agent's restart-counter probe and its saved state add eight frozen

// declarations.

- seal = "edaf4676482aca981d259d4e574a09d53ef5f8330cbcc9bfa69057c644426ca5"

+ seal = "bb7e060d46e881784ee3373af99ab3b714749c47e2960e914066d7b1fe0a6286"

errs := append(compareRootReads(got, inv), checkRootReadLegacy(inv, seal)...)

errs = append(errs, checkRootReadDeclarations(got, inv)...)

if len(errs) > 0 {

diff --git a/internal/provision/testdata/plan14-root-reads-ls284.tsv b/internal/provision/testdata/plan14-root-reads-ls284.tsv

index 5ac00676f..720ed4871 100644

--- a/internal/provision/testdata/plan14-root-reads-ls284.tsv

+++ b/internal/provision/testdata/plan14-root-reads-ls284.tsv

@@ -8,10 +8,12 @@ internal/host/restartwatch.go loadRestartWatch legacy accepted-legacy accepted-l

internal/host/restartwatch.go platformUnitRestarts legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=The agent, as carlos, asks systemd for the restart counters of loaded carlos-*.service units; a fixed command with no registry or tenant input, the same class as internal/host/probes.go adoptTimerActive under J4 root-input. Feeds an alert only; no root decision. file:CommandContext@166 fde6a651c0a8a66c324124a2bd3085959c9b575cae08e2d65d2eb666761b0dcc

@legacy e39e5045ef43883a78c327f56fbd90e501ff3c012c8ddf512b0f7585b75623c1 internal/host/units/carlos-litestream.service external legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Memory cap default 512M with GOMEMLIMIT, and restart backoff (RestartSteps, RestartMaxDelaySec). User, paths, command and sandboxing unchanged; no new source or authority. file:external@0 2fc82f963202103a16a89aaecdcaeb89c6b9fee8e9d0e09cb76389a2326c1714

internal/host/units/carlos-litestream.service external legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Memory cap default 512M with GOMEMLIMIT, and restart backoff (RestartSteps, RestartMaxDelaySec). User, paths, command and sandboxing unchanged; no new source or authority. file:external@0 2fc82f963202103a16a89aaecdcaeb89c6b9fee8e9d0e09cb76389a2326c1714

-@legacy 8d4ad62d122c3adc99b4ac55ff45d965ec1982a024245825a2a37dd82acec4e1 internal/ops/litestream_memory.go writeMemoryDropIn legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root opens the drop-in's own directory to hold a lock, reads the root-owned drop-in this verb itself wrote, and lstats its own reload-pending marker beside it, to decide whether to rewrite and reload; the same compare-before-write as writeIfChanged under J4 root-input. The path is fixed or an operator-supplied flag. file:Lstat@613,file:Open@264,file:ReadFile@454 9b0fb46e224f97d4740165498abe45a1c25004a955c6176bce20b08398b46f7c

-internal/ops/litestream_memory.go writeMemoryDropIn legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root opens the drop-in's own directory to hold a lock, reads the root-owned drop-in this verb itself wrote, and lstats its own reload-pending marker beside it, to decide whether to rewrite and reload; the same compare-before-write as writeIfChanged under J4 root-input. The path is fixed or an operator-supplied flag. file:Lstat@613,file:Open@264,file:ReadFile@454 9b0fb46e224f97d4740165498abe45a1c25004a955c6176bce20b08398b46f7c

-@legacy dc2175dd0445552045146497dd473327ec0ddca2de728f6b8a2c5c755b82420d internal/ops/ops.go runLitestream legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Same registry open and prior-config read. Adds one root-owned write, the memory drop-in beside the unit, sized from the database count this run computed, and a daemon-reload; no new input is trusted. data-literal@214,file:Open@1383,file:ReadFile@2357 380a01961cc02f2a18673bc4ad0a0bc1112905d4726f8ab9e7b370da36686306

-internal/ops/ops.go runLitestream legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Same registry open and prior-config read. Adds one root-owned write, the memory drop-in beside the unit, sized from the database count this run computed, and a daemon-reload; no new input is trusted. data-literal@214,file:Open@1383,file:ReadFile@2357 380a01961cc02f2a18673bc4ad0a0bc1112905d4726f8ab9e7b370da36686306

+@legacy 23e4a68a4a7c5c6ef7a0221cab2c4da82f6318dd1461d0f61b00e21638847f7c internal/ops/litestream_memory.go configDBCount legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root counts the databases in the generated config this verb itself just wrote, under the drop-in lock, to size the memory cap; the config is root-owned and is the same file RestoreVerify reads under J4 root-input. file:ReadFile@61 6d3fc5e14582c88f3e2198858fd5b8c4d2b2cbb96c54d850effacde083a56dbc

+internal/ops/litestream_memory.go configDBCount legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root counts the databases in the generated config this verb itself just wrote, under the drop-in lock, to size the memory cap; the config is root-owned and is the same file RestoreVerify reads under J4 root-input. file:ReadFile@61 6d3fc5e14582c88f3e2198858fd5b8c4d2b2cbb96c54d850effacde083a56dbc

+@legacy c07846a7bb455d43f5dd2304b4126af8583a5fbb7c6e4f788bc3d6fffa4cd40b internal/ops/litestream_memory.go writeMemoryDropIn legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root opens the drop-in's own directory to hold a lock, reads the root-owned drop-in this verb itself wrote, and lstats its own reload-pending marker beside it, to decide whether to rewrite and reload; the same compare-before-write as writeIfChanged under J4 root-input. The path is fixed or an operator-supplied flag. file:Lstat@713,file:Open@259,file:ReadFile@551 32347ee060400b8353b0f00924cd62024c1df1f938bd3ec06830dfbf58e64f61

+internal/ops/litestream_memory.go writeMemoryDropIn legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root opens the drop-in's own directory to hold a lock, reads the root-owned drop-in this verb itself wrote, and lstats its own reload-pending marker beside it, to decide whether to rewrite and reload; the same compare-before-write as writeIfChanged under J4 root-input. The path is fixed or an operator-supplied flag. file:Lstat@713,file:Open@259,file:ReadFile@551 32347ee060400b8353b0f00924cd62024c1df1f938bd3ec06830dfbf58e64f61

+@legacy 2057fce9777b5b8c36dfd99be6b8f96ffdc488c8872fa6cc777857c5d1bc9306 internal/ops/ops.go runLitestream legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Same registry open and prior-config read. Adds one root-owned write, the memory drop-in beside the unit, sized from the database count this run computed, and a daemon-reload; no new input is trusted. data-literal@214,file:Open@1383,file:ReadFile@2357 b55ef3cb905a4d5f34b19c70e35a0e837258c379f7405d90a8a183311f2afc2c

+internal/ops/ops.go runLitestream legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Same registry open and prior-config read. Adds one root-owned write, the memory drop-in beside the unit, sized from the database count this run computed, and a daemon-reload; no new input is trusted. data-literal@214,file:Open@1383,file:ReadFile@2357 b55ef3cb905a4d5f34b19c70e35a0e837258c379f7405d90a8a183311f2afc2c

@legacy e2fa62786280b07144aec18b9255bcafb35176e10ca2f6467cfc1656691695cb internal/ops/ops.go systemdDaemonReload legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root runs a fixed systemctl daemon-reload with no caller input, the same class as internal/host/bootstrap.go systemctl under J4 root-input. file:CommandContext@143 42322a947e1763aa00af63d28155b0919cf72ec9909d49e4ec860cfc7817fd4e

internal/ops/ops.go systemdDaemonReload legacy accepted-legacy accepted-legacy;rev3=census-ruling-ls284;basis=Root runs a fixed systemctl daemon-reload with no caller input, the same class as internal/host/bootstrap.go systemctl under J4 root-input. file:CommandContext@143 42322a947e1763aa00af63d28155b0919cf72ec9909d49e4ec860cfc7817fd4e

@coverage internal/console 85d4f9521a2acc08aa1276f414be41cea32aa4275f761546e77e951d0d58b558

@@ -19,4 +21,4 @@ internal/ops/ops.go systemdDaemonReload legacy accepted-legacy accepted-legacy;r

@coverage internal/host 8bf2f3b97227361f378bbd8943d2d0e157a66f6f90baa5291fb8a13649d4e7de

@coverage internal/host/units e06e4a1a0d9d45f8e7494f0419834bf23e14fb2cd4861b94ef905bfaf795a7d1

@coverage internal/metrics 1adc51fc485ba895970c22277b7bff997c24ebededbd19f48a10e75cf531323a

-@coverage internal/ops 1ca0655a6fe33c03993a55c7c08689f8fc1b7adbba14efe5eead823657713353

+@coverage internal/ops acd1cb6aced4557a7836f4d69f62e69599fabf94ef8e5b5951480efb8b8e7273