amadan

carlos / platform Public

moves: one lock per move journal, shared by every writer

moves-phase1-6 Tasks 4+5 fix round 1, ruling 3. Every write of a move's
journal holds that journal's one lock (moves.journalLock, per journal
directory), and under it keeps the steps another writer added since
this value was read; a step the writer read and dropped stays dropped.
moves.UpdateJournal holds the lock across the read too, and the
landing's admitting mark (moveslandprobe.go), recordFilesDigest and the
paging rules use it. pageStartRefused no longer takes the owners' lock.

Tests: TestJournalWritersKeepEachOthersSteps (a stale writer keeps the
other's step, both orders) and a deterministic interleave in
TestPagingIsRateLimited: a start check page that comes while the
landing holds the journal waits, and both marks survive.

No censused unit changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4002bd1e8a2bbfaa9a655f85c5c1caf5beeb737d parent e8827f7
6 files changed, +224 −39
  • internal/host/moveslandprobe.go +7 −6
  • internal/host/movespaging.go +28 −23
  • internal/host/movespaging_test.go +38 −0
  • internal/host/movesprobeowners.go +17 −9
  • internal/moves/journal.go +76 −1
  • internal/moves/journal_lock_test.go +58 −0
diff --git a/internal/host/moveslandprobe.go b/internal/host/moveslandprobe.go
index 60acc0ce6..a379c880a 100644
--- a/internal/host/moveslandprobe.go
+++ b/internal/host/moveslandprobe.go
@@ -313,12 +313,13 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) {
if err != nil {
return false, err
}
- if j, err = l.journal(); err != nil {
- return false, err
- }
- j.Phase, j.Attempt = provision.LandProbing, k
- j.Mark(probeStepAdmitting+strconv.Itoa(k), string(admitting))
- if err := j.Write(cfg.DataDir); err != nil {
+ // Under the journal's one lock (moves.UpdateJournal): the probe
+ // owners and the paging rules write this journal too.
+ if err := moves.UpdateJournal(cfg.DataDir, l.rec.Host, l.g.ID, func(j *moves.Journal) error {
+ j.Phase, j.Attempt = provision.LandProbing, k
+ j.Mark(probeStepAdmitting+strconv.Itoa(k), string(admitting))
+ return nil
+ }); err != nil {
return false, err
}
if err := l.fault("before-probe-" + strconv.Itoa(k)); err != nil {
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go
index c196c7ed7..c3c3b11c8 100644
--- a/internal/host/movespaging.go
+++ b/internal/host/movespaging.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "errors"
"fmt"
"log"
"strconv"
@@ -40,7 +41,8 @@ import (
//
// A rule's rate state is the op journal on this box (moves.Journal, step
// paged:<rule>, the time it was marked), so a restarted agent does not page
-// it again. The step is written before the page is sent: a crash between the
+// it again. It is read and written under the journal's one lock
+// (moves.UpdateJournal), which every other writer of the journal shares. The step is written before the page is sent: a crash between the
// two, or a send that fails (logged), pages nothing more, as the console's
// move_pages records an attempt whether or not delivery worked (pages are
// best effort). A journal that cannot be read or written sends nothing, so a
@@ -105,16 +107,17 @@ func pageOnce(ctx context.Context, page func(ctx context.Context, level, subject
unwiredOnce(dataDir, host, op, rule, now)
return pageUnwired
}
- j, err := moves.ReadJournal(dataDir, host, op)
- if err != nil {
- pagingLogf("moves: %s: the %s page of %s is not sent: its journal: %v", host, rule, op, err)
- return pageUndelivered
- }
- if j.Done(pagedStep + rule) {
+ err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error {
+ if j.Done(pagedStep + rule) {
+ return errJournalUnchanged
+ }
+ j.Mark(pagedStep+rule, now.UTC().Format(time.RFC3339))
+ return nil
+ })
+ if errors.Is(err, errJournalUnchanged) {
return pagePagedBefore
}
- j.Mark(pagedStep+rule, now.UTC().Format(time.RFC3339))
- if err := j.Write(dataDir); err != nil {
+ if err != nil {
pagingLogf("moves: %s: the %s page of %s is not sent: its journal: %v", host, rule, op, err)
return pageUndelivered
}
@@ -130,18 +133,23 @@ func pageOnce(ctx context.Context, page func(ctx context.Context, level, subject
// step remembers. A journal that cannot be read or written does not quiet
// it: the warning is logged again the next time.
func unwiredOnce(dataDir, host, op, rule string, now time.Time) {
- j, err := moves.ReadJournal(dataDir, host, op)
- if err == nil && j.Done(unwiredStep+rule) {
- return
- }
- pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op)
- if err != nil {
+ warned := false
+ err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error {
+ if j.Done(unwiredStep + rule) {
+ return errJournalUnchanged
+ }
+ warned = true
+ pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op)
+ j.Mark(unwiredStep+rule, now.UTC().Format(time.RFC3339))
+ return nil
+ })
+ if err == nil || errors.Is(err, errJournalUnchanged) {
return
}
- j.Mark(unwiredStep+rule, now.UTC().Format(time.RFC3339))
- if err := j.Write(dataDir); err != nil {
- pagingLogf("moves: %s: the warning about the unwired %s page of %s is not recorded, so it is logged again: %v", host, rule, op, err)
+ if !warned {
+ pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op)
}
+ pagingLogf("moves: %s: the warning about the unwired %s page of %s is not recorded, so it is logged again: %v", host, rule, op, err)
}
// pageOnce is the worker's pageOnce: its pager, journals and clock.
@@ -203,17 +211,14 @@ func (w *movesWorker) pageStray(ctx context.Context, rec provision.Record) {
}
// pageStartRefused pages an incident once per op for a refused §5.1 start
-// check on this box (§13), and returns what became of the page. The probe
-// owners' lock is held across the journal's read and write, as their other
-// journal writes hold it.
+// check on this box (§13), and returns what became of the page. Its journal
+// write takes the journal's one lock, which the landing's pass shares.
func (m *movesProbeOwners) pageStartRefused(ctx context.Context, host, op string) pageOutcome {
now := time.Now()
if m.Now != nil {
now = m.Now()
}
subject, body := startRefusedPage(host, op, m.Box)
- m.mu.Lock()
- defer m.mu.Unlock()
return pageOnce(ctx, m.Page, m.DataDir, host, op, pageRuleStartCheck, now, alerting.LevelIncident, subject, body)
}
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go
index eeed1db75..67ab4b747 100644
--- a/internal/host/movespaging_test.go
+++ b/internal/host/movespaging_test.go
@@ -273,6 +273,44 @@ func TestPagingIsRateLimited(t *testing.T) {
}
})
+ // Fix round 1, ruling 3: one lock per journal. A refused start check's
+ // page that comes while the landing holds its journal (moveslandprobe's
+ // admitting mark, through moves.UpdateJournal) waits for it, and both
+ // the landing's mark and paged:start_check are in the journal after.
+ t.Run("one journal lock: the landing's mark and the page both survive", func(t *testing.T) {
+ f := newPOFixture(t)
+ var pages pagesSeen
+ f.owners.Page = pages.hook()
+ done := make(chan pageOutcome, 1)
+ err := moves.UpdateJournal(f.dataDir, poHost, poOp, func(j *moves.Journal) error {
+ go func() { done <- f.owners.pageStartRefused(ctx, poHost, poOp) }()
+ select {
+ case o := <-done:
+ t.Errorf("the page wrote the journal (%s) while the landing held it", o)
+ case <-time.After(100 * time.Millisecond):
+ }
+ j.Phase, j.Attempt = provision.LandProbing, 1
+ j.Mark(probeStepAdmitting+"1", `{"session":7}`)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o := <-done; o != pageSent {
+ t.Fatalf("the page after the landing's write: %s", o)
+ }
+ j, err := moves.ReadJournal(f.dataDir, poHost, poOp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !j.Done(probeStepAdmitting+"1") || !j.Done(pagedStep+pageRuleStartCheck) || j.Phase != provision.LandProbing {
+ t.Fatalf("the journal lost a writer's step: %+v", j)
+ }
+ if n, _ := pagesWith(pages, "carlos move start refused: "); n != 1 {
+ t.Fatalf("pages %+v, want one", pages)
+ }
+ })
+
t.Run("a journal that cannot be written sends nothing", func(t *testing.T) {
var pages pagesSeen
if pageOnce(ctx, pages.hook(), t.TempDir(), "bad/host", p1ID, pageRuleStray, time.Now(), "incident", "s", "b") == pageSent || len(pages) != 0 {
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index 7cee17027..fb84a9a41 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -1111,20 +1111,28 @@ func (m *movesProbeOwners) filesDigest(host string) (string, error) {
func (m *movesProbeOwners) recordFilesDigest(host, op, g, digest string) error {
m.mu.Lock()
defer m.mu.Unlock()
- j, err := moves.ReadJournal(m.DataDir, host, op)
- if err != nil {
- return err
- }
- if have, ok := j.Steps[filesPlainStep+g]; ok {
- if have != digest {
- return fmt.Errorf("%w: generation %s recorded other files", hostbox.ErrChanged, g)
+ // Under the journal's one lock (moves.UpdateJournal): the landing's pass
+ // and the paging rules write this journal too.
+ err := moves.UpdateJournal(m.DataDir, host, op, func(j *moves.Journal) error {
+ if have, ok := j.Steps[filesPlainStep+g]; ok {
+ if have != digest {
+ return fmt.Errorf("%w: generation %s recorded other files", hostbox.ErrChanged, g)
+ }
+ return errJournalUnchanged
}
+ j.Mark(filesPlainStep+g, digest)
+ return nil
+ })
+ if errors.Is(err, errJournalUnchanged) {
return nil
}
- j.Mark(filesPlainStep+g, digest)
- return j.Write(m.DataDir)
+ return err
}
+// errJournalUnchanged ends a moves.UpdateJournal that has nothing to write;
+// its caller treats it as success.
+var errJournalUnchanged = errors.New("moves journal: nothing to write")
+
// recordedFilesDigest is generation g's recorded files digest: the move id's
// is the source's stopped baseline the landing installed (Move.Source.Files);
// a later generation's is the one its publication journalled. false when
diff --git a/internal/moves/journal.go b/internal/moves/journal.go
index d83fe8fef..d0a3410a3 100644
--- a/internal/moves/journal.go
+++ b/internal/moves/journal.go
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync"
"time"
)
@@ -31,6 +32,56 @@ type Journal struct {
Steps map[string]string `json:"steps,omitempty"`
Attempt int `json:"attempt,omitempty"`
UpdatedAt time.Time `json:"updated_at"`
+
+ // read is the steps this value held when it was read or last written:
+ // a step on disk that it does not hold and did not read was another
+ // writer's, and Write keeps it.
+ read map[string]bool
+}
+
+// A move's journal has more than one writer on a box: the pass that runs the
+// phase, the probe owners and attempts, the paging rules (Plan 6 fix round
+// 1, ruling 3). Every write of one journal holds that journal's lock, the one
+// lock every writer in this process shares (journalLock), and under it keeps
+// any step another writer added since this value was read, so no writer's
+// step is lost to another's read-modify-write. UpdateJournal holds the lock
+// across the read as well, for a writer whose decision depends on what is
+// there (a rate limit, a recorded digest).
+var journalLocks sync.Map // the journal's directory -> *sync.Mutex
+
+// journalLock is the lock of the journal of (host, op) under dataDir.
+func journalLock(dataDir, host, op string) *sync.Mutex {
+ l, _ := journalLocks.LoadOrStore(filepath.Clean(JournalDir(dataDir, host, op)), &sync.Mutex{})
+ return l.(*sync.Mutex)
+}
+
+// UpdateJournal reads the journal of (host, op), calls fn with it and writes
+// it, all under the journal's lock. An error from fn writes nothing and is
+// returned. fn must not call Write.
+func UpdateJournal(dataDir, host, op string, fn func(*Journal) error) error {
+ if err := journalNames(host, op); err != nil {
+ return err
+ }
+ l := journalLock(dataDir, host, op)
+ l.Lock()
+ defer l.Unlock()
+ j, err := ReadJournal(dataDir, host, op)
+ if err != nil {
+ return err
+ }
+ if err := fn(j); err != nil {
+ return err
+ }
+ return j.write(dataDir)
+}
+
+// readSteps is the set of steps j holds now.
+func (j *Journal) readSteps() map[string]bool {
+ out := make(map[string]bool, len(j.Steps))
+ for k := range j.Steps {
+ out[k] = true
+ }
+ return out
}
// JournalDir is where one operation's journal lives on the box:
@@ -89,16 +140,39 @@ func ReadJournal(dataDir, host, op string) (*Journal, error) {
if j.Steps == nil {
j.Steps = map[string]string{}
}
+ j.read = j.readSteps()
return &j, nil
}
// Write stamps UpdatedAt and saves the journal: a temp file in the same
// directory, fsync, rename over the old file, then fsync of the directory,
-// so a crash leaves either the old journal or the new one.
+// so a crash leaves either the old journal or the new one. It holds the
+// journal's lock, and keeps the steps other writers added since j was read.
func (j *Journal) Write(dataDir string) error {
if err := journalNames(j.Host, j.Op); err != nil {
return err
}
+ l := journalLock(dataDir, j.Host, j.Op)
+ l.Lock()
+ defer l.Unlock()
+ return j.write(dataDir)
+}
+
+// write is Write under the journal's lock, which the caller holds. A step on
+// disk that j neither holds nor read is another writer's and is kept; a step
+// j read and then dropped stays dropped. A journal on disk that cannot be
+// read is replaced, as before.
+func (j *Journal) write(dataDir string) error {
+ if err := journalNames(j.Host, j.Op); err != nil {
+ return err
+ }
+ if disk, err := ReadJournal(dataDir, j.Host, j.Op); err == nil {
+ for k, v := range disk.Steps {
+ if _, have := j.Steps[k]; !have && !j.read[k] {
+ j.Mark(k, v)
+ }
+ }
+ }
if j.V == 0 {
j.V = journalVersion
}
@@ -150,6 +224,7 @@ func (j *Journal) Write(dataDir string) error {
os.Remove(tmpName)
return err
}
+ j.read = j.readSteps()
return syncDir(dir)
}
diff --git a/internal/moves/journal_lock_test.go b/internal/moves/journal_lock_test.go
new file mode 100644
index 000000000..dd51e629e
--- /dev/null
+++ b/internal/moves/journal_lock_test.go
@@ -0,0 +1,58 @@
+package moves
+
+import (
+ "testing"
+)
+
+// Fix round 1, ruling 3: a move's journal has more than one writer on a box
+// (the landing's pass, the probe owners, the paging rules), each a
+// read-modify-write. One shared per-journal lock serializes them, and a
+// writer whose read came before another's write keeps that write's new
+// steps: the landing's progress mark and paged:start_check both survive,
+// in either order. A step the writer itself read and deleted stays deleted.
+func TestJournalWritersKeepEachOthersSteps(t *testing.T) {
+ dd := t.TempDir()
+ read := func() *Journal {
+ t.Helper()
+ j, err := ReadJournal(dd, "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz")
+ if err != nil {
+ t.Fatal(err)
+ }
+ return j
+ }
+ write := func(j *Journal) {
+ t.Helper()
+ if err := j.Write(dd); err != nil {
+ t.Fatal(err)
+ }
+ }
+ landing, paging := read(), read()
+ paging.Mark("paged:start_check", "2026-10-11T09:00:00Z")
+ write(paging)
+ landing.Phase, landing.Attempt = "probing", 1
+ landing.Mark("probe-admitting-1", `{"session":7}`)
+ write(landing)
+ got := read()
+ if !got.Done("paged:start_check") || !got.Done("probe-admitting-1") || got.Phase != "probing" || got.Attempt != 1 {
+ t.Fatalf("after the landing's later write: %+v", got)
+ }
+
+ // The other order: the page's write lands last.
+ landing, paging = read(), read()
+ landing.Mark("probe-passed-1", "ok")
+ write(landing)
+ paging.Mark("paged:stray", "2026-10-11T09:01:00Z")
+ write(paging)
+ got = read()
+ if !got.Done("probe-passed-1") || !got.Done("paged:stray") || !got.Done("probe-admitting-1") || !got.Done("paged:start_check") {
+ t.Fatalf("after the page's later write: %+v", got.Steps)
+ }
+
+ // A step the writer read and deleted is not brought back.
+ j := read()
+ delete(j.Steps, "probe-passed-1")
+ write(j)
+ if got := read(); got.Done("probe-passed-1") || !got.Done("paged:stray") {
+ t.Fatalf("a deleted step came back, or another went: %+v", got.Steps)
+ }
+}