Heartbeat from a child process, and commit only a key still held (0.6.8)

Project.toml

@@ -1,6 +1,6 @@
 name = "SweepRunner"
 uuid = "be946ad2-3cb3-4b6e-8f7e-4a5ecc3c255b"
-version = "0.6.7"
+version = "0.6.8"
 authors = ["sota shimozono <shimozono-sota631@g.ecc.u-tokyo.ac.jp>"]
 
 [deps]
@@ -23,8 +23,9 @@ Aqua = "0.8"
 # there, and `save!` returning the digest it wrote (with `mark_done!(…; result)`) in 0.8.3. `artifact!` / `ArtifactBusy`
 # (what the deferral and `artifact_affinity` are built for) came in 0.8.2. The owner-stamped
 # `.running` API (`new_owner_token`, `running_owner`, and the three-argument `refresh_running!` /
-# `clear_running!`) that the liveness reaper is built on arrived in 0.8.1.
-DataVault = "0.8.4"
+# `clear_running!`) that the liveness reaper is built on arrived in 0.8.1. 0.8.9 is the floor now:
+# the key heartbeat is `start_heartbeat`'s child process, and a commit is `mark_done!(…, owner)`.
+DataVault = "0.8.9"
 Dates = "1.11"
 Distributed = "1.11"
 JLD2 = "0.6"

docs/src/architecture.md

@@ -79,8 +79,10 @@ When you call `run!(work_fn, vault, keys)`, it does:
      — another master may have finished this key between our manifest
      read and lock acquisition.
    - Call `work_fn(key)` up to `opts.max_attempts` times, with a heartbeat
-     task refreshing `.running`. On success, `DataVault.save!` +
-     `DataVault.mark_done!` + `Manifest.add_complete!`.
+     child process (`DataVault.start_heartbeat`) refreshing `.running`. On
+     success, `DataVault.save!` + `DataVault.mark_done!(vault, key, owner)` +
+     `Manifest.add_complete!`; the owner form commits nothing if a sibling
+     reclaimed the key meanwhile.
 5. `save_manifest(manifest)`.
 6. Emit `:stage_done` and return the aggregate counts.
 
@@ -100,11 +102,12 @@ moves on — no blocking, no waiting, no central queue.
 
 Two things keep this robust against crashes:
 
-1. **Heartbeat + stale reclaim.** The live holder touches `heartbeat` on
-   a timer. If a holder dies, its `heartbeat` mtime stops advancing; the
-   next contender sees the lock as stale (by either heartbeat age or lock-dir
-   age, depending on whether the holder made it past initial write), and
-   reclaims via `mv lock lock.dead.X` + `rm -rf lock.dead.X`.
+1. **Heartbeat + stale reclaim.** A child process of the live holder
+   rewrites the lock's `heartbeat_unix=` every `heartbeat_interval`, for as
+   long as the holder's pid lives — also while `work_fn` never yields, which
+   an in-process task did not survive. If the holder dies, the heartbeat
+   stops; after `stale_after` the next contender reclaims the lock under
+   DataVault's reclaim mutex (see DataVault's README, "ロックの規約").
 2. **Post-lock `is_done` re-check.** Even on the happy path, two masters
    can start the loop with overlapping `todo`. The re-check inside the
    locked critical section ensures the second master notices the work

src/Run.jl

@@ -31,9 +31,11 @@ Execution options for [`run!`](@ref).
   retry (a failed `work_fn` is logged as `:error` instead of `:gave_up`).
 - `stale_after::Float64 = 600.0` — seconds before another master can reclaim a
   held lock as stale. Passed through to `DataVault.acquire_running!`.
-- `heartbeat_interval::Float64 = 60.0` — how often the per-lock heartbeat task
-  refreshes DataVault's `.running` file. Enforced to be `<` `stale_after`
-  (otherwise a live holder's lock could be reclaimed mid-work).
+- `heartbeat_interval::Float64 = 60.0` — how often the per-lock heartbeat
+  (a child process, `DataVault.start_heartbeat`) refreshes DataVault's
+  `.running` file. It keeps beating while `work_fn` computes without yielding.
+  Enforced to be `<` `stale_after` (otherwise a live holder's lock could be
+  reclaimed mid-work).
 - `log_level::Symbol = :info` — event-log verbosity. At `:info` (default) the
   high-churn per-key `:lock_busy` and `:key_start` events are suppressed (their
   totals still ride in the `:stage_done` summary), keeping the JSONL log
@@ -415,7 +417,7 @@ end
 
 Execute the per-key pipeline: atomic-acquire via
 `DataVault.acquire_running!`, re-check completion, run work_fn with a
-background heartbeat task, and release on exit.  Returns a
+heartbeat child process, commit owner-checked, and release on exit.  Returns a
 `(key, outcome)` pair suitable for aggregation by the caller.
 
 Outcome symbols:
@@ -472,60 +474,25 @@ function _run_one_with_lock!(
         return (key, :already_done)
     end
 
-    # Background heartbeat task: periodically refresh `.running` so
-    # `DataVault.cleanup_stale` / sibling `acquire_running!` callers
-    # recognise us as alive.  The `Threads.Atomic{Bool}` flag makes
-    # the stop signal thread-safe; a short sleep tick keeps finally
-    # cleanup responsive (the earlier fixed 60-s sleep would block the
-    # whole shutdown until the next heartbeat tick).
-    # `lost[]` is raised by the heartbeat task if it observes that a sibling reclaimed our lock (we
-    # stalled past `stale_after`). `_run_one_with_retry!` checks it before `save!`, and the
-    # `finally` below skips `clear_running!` when lost, so we neither commit on top of nor delete
-    # the lock now owned by the reclaiming master. The refresh is OWNER-CHECKED (DataVault 0.8.1),
-    # so a reclaim that has already happened is seen on the next beat; the previous
-    # existence-based form returned `true` against the reclaimer's own file.
-    hb_stop = Threads.Atomic{Bool}(false)
-    lost = Threads.Atomic{Bool}(false)
-    hb_task = Threads.@spawn begin
-        tick = 0.1
-        elapsed = 0.0
-        while !hb_stop[]
-            sleep(tick)
-            hb_stop[] && break
-            elapsed += tick
-            if elapsed >= opts.heartbeat_interval
-                # A `false` return (sibling reclaimed) OR a throw (un-refreshable
-                # lock, e.g. NFS hiccup) both mean "treat as lost": stop
-                # heartbeating and signal it, rather than silently dying.
-                alive = try
-                    DataVault.refresh_running!(vault, key, tok)
-                catch
-                    false
-                end
-                if !alive
-                    lost[] = true
-                    break
-                end
-                elapsed = 0.0
-            end
-        end
-    end
+    # The heartbeat runs in a CHILD process (DataVault 0.8.9). A task here was starved by work that
+    # does not yield — a long BLAS call, a tight loop — under -t 1, -t 2 and -t 2,1 alike: it never
+    # beat, a sibling reclaimed the live key after `stale_after`, and this master committed it too.
+    # The child stops when this process dies or when the lock leaves our hands.
+    #
+    # Whether we still hold the key is asked where it matters, at commit: `_run_one_with_retry!`
+    # checks the owner before `save!`, and commits with the owner form of `mark_done!`, which
+    # refuses if a sibling reclaimed in between. The release below is owner-checked as well, so it
+    # can run unconditionally: it never deletes a reclaimer's lock.
+    hb = DataVault.start_heartbeat(vault, key, tok; interval=opts.heartbeat_interval)
 
     outcome = try
-        _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, lost)
+        _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, tok)
     finally
-        hb_stop[] = true
-        try
-            wait(hb_task)
-        catch
-        end
-        # Release the lock so the key is immediately retriable by a sibling
-        # without waiting for `stale_after` — BUT NOT if we lost it. When
-        # `lost[]`, the `.running` file is now the RECLAIMING master's, and
-        # `clear_running!` is owner-blind (`isfile && rm`), so clearing it would
-        # delete THEIR lock and re-open double-execution. On `:ok`, `mark_done!`
-        # already removed our `.running`; `clear_running!` is otherwise idempotent.
-        lost[] || DataVault.clear_running!(vault, key, tok)
+        DataVault.stop_heartbeat(hb)
+        # Release so a sibling can retry the key at once instead of after `stale_after`. On `:ok`
+        # the commit already released it; on a lost key the lock is the reclaimer's, and this is
+        # a no-op.
+        DataVault.clear_running!(vault, key, tok)
     end
 
     return (key, outcome)
@@ -773,14 +740,15 @@ function _redispatch_deferred(
 end
 
 """
-    _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, lost) -> Symbol
+    _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, tok) -> Symbol
 
 Execute `work_fn(key)` up to `opts.max_attempts` times. Returns:
   :ok        — payload saved and mark_done! called
   :gave_up   — all attempts failed, final `:gave_up` event logged
   :error     — single-attempt config (`max_attempts == 1`) that failed once
-  :lock_busy — `lost[]` was set (a sibling reclaimed our lock); the result is
-               discarded before `save!` so the reclaiming master's result wins
+  :lock_busy — the lock is no longer `tok`'s (a sibling reclaimed it): the result is
+               discarded, before `save!` if the loss is already visible, else at the
+               owner-checked commit, so the reclaiming master's result wins
 """
 function _run_one_with_retry!(
     work_fn,
@@ -790,7 +758,7 @@ function _run_one_with_retry!(
     stage::Symbol,
     log::EventLog,
     opts::RunOpts,
-    lost::Threads.Atomic{Bool},
+    tok::AbstractString,
 )
     last_err = nothing
     for attempt in 1:opts.max_attempts
@@ -802,7 +770,7 @@ function _run_one_with_retry!(
                 "work_fn must return a Dict (got $(typeof(payload))). " *
                 "Wrap scalars as e.g. Dict(\"value\" => x).",
             )
-            if lost[]
+            if DataVault.running_owner(vault, key) != tok
                 # A sibling master reclaimed our lock while work_fn ran; it now
                 # owns this key. Discard our result rather than double-committing.
                 log_event(log, :lock_lost; stage=stage, key=kstr, attempt=attempt)
@@ -811,9 +779,15 @@ function _run_one_with_retry!(
             # The digest save! took before its rename goes into the marker, so `.done` names the
             # bytes this attempt wrote rather than whatever the file holds when someone looks.
             saved = DataVault.save!(vault, key, payload)
-            DataVault.mark_done!(
-                vault, key; result=saved, observation=_observation_token(vault)
+            # Owner-checked: a reclaim between the check above and here is refused, and nothing is
+            # committed. The file save! wrote is then the reclaimer's to overwrite.
+            committed = DataVault.mark_done!(
+                vault, key, tok; result=saved, observation=_observation_token(vault)
             )
+            if !committed
+                log_event(log, :lock_lost; stage=stage, key=kstr, attempt=attempt)
+                return :lock_busy
+            end
             log_event(
                 log,
                 :key_done;

test/run/test_double_run_safety.jl

@@ -54,12 +54,14 @@ end
     try
         v = DataVault.Vault(FIXTURE_CFG; run="phase1", outdir=outdir)
         key = ParamIO.expand(v.spec)[1]
-        lost = Threads.Atomic{Bool}(true)        # simulate: heartbeat saw a reclaim
+        # Simulate a reclaim: the lock on disk belongs to a sibling, not to `tok`.
+        tok = DataVault.new_owner_token()
+        @test DataVault.acquire_running!(v, key, DataVault.new_owner_token()) === :ok
         log = SweepRunner.EventLog(joinpath(outdir, "ev.jsonl"))
         ran = Ref(false)
         wf = k -> (ran[]=true; Dict{String,Any}("x" => 1))
         outcome = SweepRunner._run_one_with_retry!(
-            wf, v, key, ParamIO.canonical(key), :phase1, log, RunOpts(), lost
+            wf, v, key, ParamIO.canonical(key), :phase1, log, RunOpts(), tok
         )
         @test ran[]                              # work_fn ran...
         @test outcome === :lock_busy             # ...but its result was discarded

test/run/test_run_adversarial.jl

@@ -10,6 +10,7 @@
 #    (exactly the situation the try-finally fix in #8 created a regression for)
 # ─────────────────────────────────────────────────────────────────────────────
 
+using Printf: @sprintf
 using SweepRunner, Test, DataVault, ParamIO, JSON3, JLD2
 using Distributed
 using Dates
@@ -90,11 +91,20 @@ end
         # exactly the crashed-run recovery path.
         DataVault.mark_running!(v, target)
         running_path = DataVault._running_file(v, target)
+        # Both heartbeat lines, as a holder that stopped beating leaves them: age is read from
+        # `heartbeat_unix=` (DataVault 0.8.9), `heartbeat=` by older readers.
         old_str = Dates.format(Dates.now() - Dates.Second(7200), "yyyy-mm-ddTHH:MM:SS")
+        old_unix = @sprintf("%017.6f", time() - 7200)
         lines = readlines(running_path)
         open(running_path, "w") do io
             for line in lines
-                println(io, startswith(line, "heartbeat=") ? "heartbeat=$old_str" : line)
+                if startswith(line, "heartbeat_unix=")
+                    println(io, "heartbeat_unix=", old_unix)
+                elseif startswith(line, "heartbeat=")
+                    println(io, "heartbeat=", old_str)
+                else
+                    println(io, line)
+                end
             end
         end
         @test DataVault.is_running(v, target)

test/run/test_run_busy_holder.jl

@@ -0,0 +1,99 @@
+# A master whose work_fn never yields keeps its key.
+#
+# The key heartbeat used to be a task in the master. Under a work_fn that does not yield — a long
+# BLAS call, a tight loop — it never beat, with -t 1, -t 2 and -t 2,1 alike; a sibling reclaimed
+# the live key after `stale_after`, and the master committed it as well. It is now DataVault's
+# heartbeat child. The master here runs in its own process (a busy loop in THIS process would
+# stop the sibling too) and the sibling polls from this one.
+
+using SweepRunner, Test, DataVault, ParamIO
+
+const _BH_CFG = joinpath(@__DIR__, "fixtures", "study.toml")
+
+function _bh_master_script(outdir, ready, busy_s)
+    return """
+    using SweepRunner, DataVault, ParamIO
+    v = DataVault.Vault($(repr(_BH_CFG)); run="busy", outdir=$(repr(outdir)))
+    k = DataVault.keys(v)[1]
+    work = key -> begin
+        touch($(repr(ready)))
+        s = let t = time() + $(busy_s), x = 0.0
+            while time() < t; x += sin(x) + 1e-9; end    # never yields
+            x
+        end
+        Dict{String,Any}("x" => s)
+    end
+    r = run!(work, v, [k]; opts=RunOpts(workers=:sequential, stale_after=1.0,
+                                         heartbeat_interval=0.2))
+    print(r.done, " ", time())
+    """
+end
+
+@testset "run!: a work_fn that never yields keeps its key against a sibling" begin
+    for threads in (1, 2)
+        outdir = mktempdir()
+        try
+            ready = joinpath(outdir, "ready")
+            out = IOBuffer()
+            exe = Base.julia_cmd()
+            script = _bh_master_script(outdir, ready, 4.0)
+            cmd = `$exe --startup-file=no -t $threads --project=$(Base.active_project()) -e $script`
+            p = run(pipeline(cmd; stdout=out, stderr=devnull); wait=false)
+            t0 = time()
+            while !isfile(ready) && process_running(p) && time() - t0 < 180
+                sleep(0.05)
+            end
+            @test isfile(ready)
+
+            v = DataVault.Vault(_BH_CFG; run="busy", outdir=outdir)
+            k = DataVault.keys(v)[1]
+            t_ready = mtime(ready)                      # when work_fn began, not when we noticed
+            got_at = Float64[]
+            tries = 0
+            while process_running(p) && !DataVault.is_done(v, k)
+                tok = DataVault.new_owner_token()
+                r = DataVault.acquire_running!(v, k, tok; stale_after=1.0)
+                tries += 1
+                if r !== :busy
+                    push!(got_at, time())
+                    DataVault.clear_running!(v, k, tok)
+                end
+                sleep(0.1)
+            end
+            wait(p)
+            done, t_end = split(String(take!(out)))
+
+            # The master really computed for several stale_after while the sibling tried.
+            @test time() - t_ready >= 3.5
+            @test tries >= 25
+            # The sibling never got the key while the master held it (only after `run!` ended).
+            @test count(<(parse(Float64, t_end)), got_at) == 0
+            @test done == "1"
+            @test DataVault.is_done(v, k)
+        finally
+            rm(outdir; recursive=true, force=true)
+        end
+    end
+end
+
+@testset "run!: a master whose key was reclaimed does not commit it" begin
+    # The loss is made to happen mid-work_fn: the sibling reclaims with stale_after=0. The master
+    # must return the key as `:lock_busy`, commit nothing, and leave the sibling's lock alone.
+    outdir = mktempdir()
+    try
+        v = DataVault.Vault(_BH_CFG; run="lost", outdir=outdir)
+        k = DataVault.keys(v)[1]
+        sib = DataVault.new_owner_token()
+        work =
+            key -> begin
+                @assert DataVault.acquire_running!(v, key, sib; stale_after=0.0) === :reclaimed
+                Dict{String,Any}("x" => 1)
+            end
+        r = run!(work, v, [k]; opts=RunOpts(workers=:sequential))
+        @test r.done == 0
+        @test !DataVault.is_done(v, k)
+        @test DataVault.running_owner(v, k) == sib
+    finally
+        rm(outdir; recursive=true, force=true)
+    end
+end

test/run/test_run_deadline.jl

@@ -52,8 +52,16 @@ end
         keys = ParamIO.expand(v.spec)
         @test length(keys) > 1
         started = Ref(0)
-        deadline = time() + 0.3
-        work = k -> (started[] += 1; sleep(0.6); Dict{String,Any}("x" => 1))
+        # The first key runs until the deadline has passed, so the test does not depend on how long
+        # `run!` takes to hand it out — only that it is under the margin. A 0.3 s margin was not:
+        # a run-start observation takes most of a second, and on a hosted runner no key was handed
+        # out at all (started == 0).
+        deadline = time() + 5.0
+        work = k -> begin
+            started[] += 1
+            sleep(max(0.0, deadline - time()) + 0.1)
+            Dict{String,Any}("x" => 1)
+        end
         r = run!(work, v, keys; opts=RunOpts(workers=:sequential, deadline=deadline))
         @test started[] >= 1                 # the first key ran
         @test started[] < length(keys)       # later keys were not handed out

test/run/test_run_loop_busy.jl

@@ -32,9 +32,12 @@ end
 
 @testset "run_loop!: a dead holder's key is waited out and completed" begin
     with_tail() do v, ks
+        # t0 BEFORE the marker, so `elapsed` bounds the marker's own age from above: the key must not
+        # be reclaimed before `stale_after` of it, and the heartbeat is exact (DataVault 0.8.9's
+        # `heartbeat_unix=`), so no allowance is due.
+        t0 = time()
         DataVault.mark_running!(v, ks[1])       # killed mid-key: the marker outlived the process
         n = Ref(0)
-        t0 = time()
         r = run_loop!(
             k -> (n[] += 1; Dict{String,Any}("x" => 1)),
             v,
@@ -48,7 +51,10 @@ end
         @test n[] == 1                          # it ran the key
         @test DataVault.is_done(v, ks[1])       # and the campaign is complete
         @test r.done == 1
-        @test elapsed >= 3.0                    # it waited stale_after out rather than giving up
+        # It waited stale_after out rather than giving up (the bug gave up after two empty rounds,
+        # ~1 s). Before DataVault 0.8.9 the heartbeat had whole seconds, the lock read up to 1 s
+        # older than it was, and this failed at 2.85 s and 2.90 s on hosted runners.
+        @test elapsed >= 3.0
     end
 end