mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
fix(checkpoint): cancel admission before invocation entry
Problem: cancelled calls remain parked behind checkpoint exclusivity, delay their provisional resource cleanup, and may later enter hooks or execution. Root cause: condition waits do not consume the invocation context. Observer cleanup also exposes a shared writer-count decrement that can run twice. Fix: centralize context-aware barrier admission, recheck cancellation after durable preparation, return idempotent permits and bind the Agent caller. Preserve completed undo retirement and every other active owner's admission. Verification: deterministic cancellation and owner-preservation schedules, compiling predecessor behavior failures, real boot direct/deferred coverage, durable undo reloads, full root inventory, owner/boot race and required suites; gofmt, vet, full lint and make check with unchanged gate baseline.
This commit is contained in:
1 parent
e59262c519
commit
ee9f75e286
10 files changed
+757
-25
No files matched your search
@@ -0,0 +1,134 @@
|
||||
---
|
||||
owner: @SivanCola
|
||||
backup: @esengine
|
||||
status: active
|
||||
reviewed: 2026-10-25
|
||||
---
|
||||
|
||||
# Checkpoint cancellation / 检查点准入取消
|
||||
|
||||
## Cause and owner
|
||||
|
||||
The checkpoint barrier waited on a condition variable without the invocation's
|
||||
context. Cancellation could leave a call waiting behind an exclusive checkpoint
|
||||
operation, then allow it to enter tool-use hooks or execution after that owner
|
||||
released. The Agent could not finish its deferred admission cleanup until then.
|
||||
|
||||
The shared barrier now accepts a context for writer and exclusive admission.
|
||||
Cancellation wakes condition waiters under the same mutex used for admission;
|
||||
the loop checks cancellation before changing ownership. A cancelled waiter does
|
||||
not decrement active writers, release exclusive access or advance generation.
|
||||
|
||||
Existing context-free methods retain their background-context behavior.
|
||||
|
||||
`MutationObserver.BeginWriteContext` uses this admission, retains the existing
|
||||
durable undo retirement, and checks cancellation again after preparation. A
|
||||
successful permit has an idempotent release closure: repeated cleanup cannot
|
||||
release a second writer.
|
||||
|
||||
The Agent passes its current invocation context after
|
||||
real target resolution and permission, before synchronous hooks or execution.
|
||||
|
||||
## Authority and data flow
|
||||
|
||||
| Boundary | Input and decision |
|
||||
| --- | --- |
|
||||
| Target and permission | Existing tool/proxy resolution and permission gates; unchanged grants |
|
||||
| Admission | Invocation context plus current barrier ownership; cancellation only abandons this request |
|
||||
| Undo preparation | Existing checkpoint Store and transaction persistence; completed retirement stays retired |
|
||||
| User code | Starts only after preparation returns a live permit; tool `Executed` is not forged |
|
||||
| Cleanup | Existing invocation defer and one permit's idempotent release |
|
||||
|
||||
The context callback only broadcasts the condition variable. It cannot grant
|
||||
admission, change another caller's ownership or interrupt an active writer.
|
||||
|
||||
No file path, command, shell launcher, approval bypass or arbitrary callback is
|
||||
introduced into the barrier. No persisted format, cross-process claim, wire
|
||||
field, tool schema, provider prefix, permission setting or default changes.
|
||||
|
||||
## Compatibility and limits
|
||||
|
||||
The barrier coordinates a Store's writers against rewind. Concurrent writers
|
||||
remain supported; this is not file-level exclusion or a cross-process lease.
|
||||
|
||||
Existing nonblocking `TryEnter*` calls keep their behavior. Production rewind
|
||||
uses these nonblocking methods; the new exclusive context method does not turn
|
||||
rewind into a queued operation or change its busy refusal.
|
||||
|
||||
Cancellation governs admission, not a revocable execution permit. Once user
|
||||
code starts, its own cancellation and physical completion rules still apply.
|
||||
|
||||
Undo persistence and ordinary mutex acquisition are not interrupted. If
|
||||
cancellation occurs during durable preparation, the completed retirement is
|
||||
preserved, this permit is released, and the call stops before user code.
|
||||
|
||||
Old undo is not restored after another action may have invalidated it.
|
||||
|
||||
The legacy background `RegisterWriter` path still uses context-free admission
|
||||
under its registry mutex. Its cancellation, per-registration identity and
|
||||
publication lifetime require a separate lifecycle change.
|
||||
|
||||
An asynchronous hook
|
||||
can outlive its synchronous callback; this slice does not register or reconcile
|
||||
that process.
|
||||
|
||||
No FIFO, bounded wait, model-capacity separation, complete effect
|
||||
receipt, safe isolated integration or whole-series completion is claimed.
|
||||
|
||||
## Reachability and validation
|
||||
|
||||
Deterministic Agent tests hold an exclusive barrier, park direct and deferred
|
||||
writer/reader-hook calls, cancel, then require completion while the owner still
|
||||
holds access. They verify no callback/tool entry or changed user bytes, and no
|
||||
leaked writer. Original owners compile and fail these behavioral assertions.
|
||||
|
||||
A live actor's aborted checkpoint wait must also release its provisional
|
||||
workspace paths so a second Owner can acquire them before the actor ends.
|
||||
|
||||
Real `boot.Build` tests retain built-in tools, capability resolution and shell
|
||||
hooks. The fixture injects a Store-bound observer through the public Agent
|
||||
binding and cancels at the permission boundary.
|
||||
|
||||
Original reader-hook cases
|
||||
remain blocked and attempt hook entry after cleanup releases exclusive access;
|
||||
the shell launcher itself refuses that cancelled command.
|
||||
|
||||
Built-in writer
|
||||
cases are controls: the existing workspace lease already refuses them first.
|
||||
|
||||
This is an assembly regression with an explicit collaborator, not evidence
|
||||
that production rewind deliberately holds the barrier indefinitely.
|
||||
|
||||
Barrier tests cover cancelled/free admission, queued writer/exclusive waits,
|
||||
deadline expiry, shutdown, successful legacy/context waiters and preservation
|
||||
of another owner and generation. Fake time and `synctest` establish ordering;
|
||||
the boot watchdog only detects a stuck call, not a performance threshold.
|
||||
|
||||
Public checkpoint APIs create a committed undo offer. Cancelling before
|
||||
admission preserves it across reload. Cancelling through the Store's undo-change
|
||||
notification after retirement keeps it retired across reload. Repeated permit
|
||||
cleanup cannot admit rewind while a second writer remains.
|
||||
|
||||
These fixtures
|
||||
exercise owner contracts rather than asserting an observed production incident.
|
||||
|
||||
Local validation and authored security analysis do not replace exact-head
|
||||
hosted platform checks or independent review.
|
||||
|
||||
## 中文说明
|
||||
|
||||
旧检查点屏障不接收调用 context,取消无法唤醒等待者;独占方稍后退出后,旧调用
|
||||
仍可能进入 hook 或工具。现在等待、取消和准入由同一屏障所有者协调,取消只撤回
|
||||
本次请求,不会释放别人的许可或推进版本。Agent 在实际目标解析和权限检查之后,
|
||||
将调用 context 传给检查点准入,成功许可仍由调用退出路径释放。
|
||||
|
||||
准入前取消保留既有 undo;持久化退役期间发生取消,则保留已经完成的退役并退出,
|
||||
不恢复可能已失效的旧 undo。重复释放同一许可不会减少其他写入者的计数。
|
||||
|
||||
普通上下文无关 API、非阻塞 rewind、并行写入、权限、持久化格式和工具定义保持
|
||||
既有约定。本次不强制打断进行中的文件 I/O,不解决后台登记锁、异步 hook 进程、
|
||||
公平队列、跨进程文件事务、验证证据或整个隔离恢复生命周期。
|
||||
|
||||
真实装配测试明确注入检查点 Store;旧读取 hook 路径会阻塞并尝试进入回调,但实际
|
||||
shell 启动器会拒绝已取消的命令,不能把它描述成已运行 shell 写入。写工具对照由
|
||||
现有工作区租约先行拒绝。确定性交错、磁盘重载、本地 race 和后续 CI 分开记账。
|
||||
@@ -27,7 +27,7 @@ checkpoint complete. Tool execution and read-only classification are unchanged.
|
||||
| --- | --- |
|
||||
| Target access | Existing permission and target resolution finish before checkpoint admission |
|
||||
| Hook capability | Existing optional capability report; legacy custom hooks remain conservatively write-capable |
|
||||
| Checkpoint owner | Existing `MutationObserver.BeginWrite`: enter rewind barrier, durably retire undo, then return permit |
|
||||
| Checkpoint owner | `MutationObserver.BeginWriteContext`: enter rewind barrier, durably retire undo, then return permit |
|
||||
| Failure | Closed barrier or failed undo retirement refuses the call before hook/tool entry |
|
||||
| Cleanup | Existing invocation defer releases the permit on success, refusal or failure |
|
||||
| Coverage | Existing `GapHookWrite`, never a fabricated path or captured preimage |
|
||||
@@ -46,9 +46,9 @@ The checkpoint barrier excludes rewind; it does not serialize all writers or
|
||||
replace file-level, cross-process mutation coordination. Readers with possible
|
||||
hook writes now follow the existing writer wait/shutdown contract.
|
||||
|
||||
The existing barrier wait has no context argument. This change does not add a
|
||||
cancellable fair queue or claim to fix its wait semantics. It introduces no
|
||||
new global mutex or filesystem scan.
|
||||
Invocation admission accepts its context; cancellation can abandon a wait
|
||||
without releasing the exclusive owner. See [cancellation bounds](CHECKPOINT_CANCELLATION_SECURITY.md).
|
||||
This does not add a fair queue, global execution mutex or filesystem scan.
|
||||
|
||||
Imported asynchronous hooks can outlive their callback. Their process lifetime
|
||||
is not covered by this permit; registering and reconciling those background
|
||||
@@ -97,8 +97,8 @@ undo,再允许回调执行。准入失败则 hook 和工具都不执行;调
|
||||
hook 的未知写入只记录既有覆盖缺口,不伪造预映像或修改工具 `Executed`。无工具
|
||||
hook 的读取保持原行为。权限、工具定义、配置和持久化格式不变。
|
||||
|
||||
该屏障只保护 checkpoint 与 rewind,不是完整文件事务或跨进程锁;原屏障等待的
|
||||
取消能力、验证证据、自动隔离与恢复仍需后续实现。测试的存储故障是明确故障注入,
|
||||
该屏障只保护 checkpoint 与 rewind,不是完整文件事务或跨进程锁;调用等待支持
|
||||
取消,但后台登记、验证证据、自动隔离与恢复仍需后续实现。测试的存储故障是明确故障注入,
|
||||
不冒称已在线上发生。
|
||||
|
||||
导入的异步 hook 进程可能晚于回调结束,当前许可不覆盖它的完整运行期;这类后台
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
package boot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/ext/hook"
|
||||
"reasonix/internal/session/control"
|
||||
"reasonix/internal/state/checkpoint"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
type checkpointCancelGate struct {
|
||||
cancel context.CancelFunc
|
||||
checked chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (g *checkpointCancelGate) Check(context.Context, string, json.RawMessage, bool) (bool, string, error) {
|
||||
g.once.Do(func() {
|
||||
g.cancel()
|
||||
close(g.checked)
|
||||
})
|
||||
return true, "", nil
|
||||
}
|
||||
|
||||
func TestEffectCancelledCheckpointAdmissionStopsUserCode(t *testing.T) {
|
||||
for _, name := range []string{"read_file", "write_file"} {
|
||||
for _, proxy := range []bool{false, true} {
|
||||
scenario := name + "/direct"
|
||||
if proxy {
|
||||
scenario = name + "/capability"
|
||||
}
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
home, root := statedBootHome(t), t.TempDir()
|
||||
writeFile(t, root, "note.txt", "user bytes")
|
||||
marker := filepath.Join(root, "hook.txt")
|
||||
args := `{"path":"note.txt"}`
|
||||
if name == "write_file" {
|
||||
args = `{"path":"note.txt","content":"tool bytes"}`
|
||||
}
|
||||
call := scriptedCall{name: name, args: args}
|
||||
if proxy {
|
||||
raw, err := json.Marshal(map[string]any{"action": "call", "capability_id": "tool:" + name, "arguments": json.RawMessage(args)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
call = scriptedCall{name: "use_capability", args: string(raw)}
|
||||
}
|
||||
kind := "checkpoint-cancel-" + strings.ReplaceAll(t.Name(), "/", "-")
|
||||
p := &scriptedCallProvider{calls: []scriptedCall{call}}
|
||||
provider.Register(kind, func(provider.Config) (provider.Provider, error) { return p, nil })
|
||||
writeFile(t, root, "reasonix.toml", `default_model = "fixture"
|
||||
[codegraph]
|
||||
enabled = false
|
||||
[[providers]]
|
||||
name = "fixture"
|
||||
kind = "`+kind+`"
|
||||
model = "fixture"
|
||||
`)
|
||||
approveWorkspaceAt(t, home, root)
|
||||
if name == "read_file" {
|
||||
settings := hook.Settings{Hooks: map[hook.Event][]hook.HookConfig{hook.PreToolUse: {{Match: name, Command: markerCommand(t, marker)}}}}
|
||||
raw, err := json.Marshal(settings)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(home, hook.SettingsFilename), raw, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
sink := &hookWorkspaceSink{}
|
||||
ctrl, err := Build(context.Background(), Options{Home: home, WorkspaceRoot: root, Sink: sink})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { ctrl.Close() })
|
||||
ctrl.SetFreshSessionPath(sessionstore.NewSessionPath(ctrl.SessionDir(), ctrl.Label()))
|
||||
ctrl.SetToolApprovalMode(control.ToolApprovalYolo)
|
||||
// The fixture supplies the checkpoint owner through the public Agent
|
||||
// binding; tools, proxy resolution and hooks still come from Build.
|
||||
store := checkpoint.New("", root)
|
||||
ctrl.Executor().SetMutationObserver(checkpoint.NewMutationObserver(checkpoint.ObserverOptions{Store: store}))
|
||||
if err := store.Barrier().EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var release sync.Once
|
||||
unlock := func() { release.Do(store.Barrier().ExitExclusive) }
|
||||
defer unlock()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
gate := &checkpointCancelGate{cancel: cancel, checked: make(chan struct{})}
|
||||
ctrl.Executor().SetGate(gate)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- ctrl.Run(ctx, "exercise checkpoint cancellation") }()
|
||||
select {
|
||||
case <-gate.checked:
|
||||
case <-time.After(10 * time.Second):
|
||||
unlock()
|
||||
cancel()
|
||||
<-done
|
||||
t.Fatal("script did not reach the permission boundary")
|
||||
}
|
||||
select {
|
||||
case err = <-done:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Error("cancelled invocation still waits for the exclusive checkpoint owner")
|
||||
unlock()
|
||||
err = <-done
|
||||
}
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("Run returned an unrelated error: %v", err)
|
||||
}
|
||||
if store.Barrier().HasWriters() {
|
||||
t.Error("cancelled invocation retained writer admission")
|
||||
}
|
||||
if _, err := os.Stat(marker); !os.IsNotExist(err) {
|
||||
t.Errorf("cancelled reader entered its hook: %v", err)
|
||||
}
|
||||
if data, err := os.ReadFile(filepath.Join(root, "note.txt")); err != nil || string(data) != "user bytes" {
|
||||
t.Errorf("cancelled invocation changed user bytes: %q %v", data, err)
|
||||
}
|
||||
sink.mu.Lock()
|
||||
defer sink.mu.Unlock()
|
||||
if len(sink.results) != 1 || sink.results[0].Executed || sink.results[0].WorkspaceMutation {
|
||||
t.Errorf("cancelled invocation's execution/effect result = %+v", sink.results)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/state/checkpoint"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
"reasonix/internal/state/workspacelease"
|
||||
)
|
||||
|
||||
func TestCancelledCheckpointWaitNeverEntersToolOrHook(t *testing.T) {
|
||||
for _, reader := range []bool{false, true} {
|
||||
for _, proxy := range []bool{false, true} {
|
||||
name := "writer"
|
||||
if reader {
|
||||
name = "reader_hook"
|
||||
}
|
||||
if proxy {
|
||||
name += "/capability"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
path := filepath.Join(root, "note.txt")
|
||||
if err := os.WriteFile(path, []byte("user bytes"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
store := checkpoint.New("", root)
|
||||
if err := store.Barrier().EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var target tool.Tool = &checkpointWriteTool{path: path}
|
||||
var hooks ToolHooks
|
||||
hookCalls := 0
|
||||
if reader {
|
||||
target = fakeTool{name: "probe", readOnly: true}
|
||||
hooks = &checkpointToolHooks{pre: func() {
|
||||
hookCalls++
|
||||
if err := os.WriteFile(path, []byte("hook bytes"), 0o600); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}}
|
||||
}
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(target)
|
||||
call := provider.ToolCall{ID: "cancelled", Name: target.Name(), Arguments: `{}`}
|
||||
if proxy {
|
||||
reg.Add(readOnlyBoundaryProxy{resolved: tool.ResolvedCall{Target: target, TargetName: target.Name(), ReadOnly: reader, Args: json.RawMessage(`{}`)}})
|
||||
call.Name = "use_capability"
|
||||
}
|
||||
a := New(nil, reg, sessionstore.NewSession(""), Options{Hooks: hooks}, event.Discard)
|
||||
a.SetMutationObserver(checkpoint.NewMutationObserver(checkpoint.ObserverOptions{Store: store}))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan toolOutcome, 1)
|
||||
go func() { done <- a.executeOne(ctx, &a.turn, call) }()
|
||||
synctest.Wait()
|
||||
select {
|
||||
case out := <-done:
|
||||
t.Fatalf("call did not wait for exclusive owner: %+v", out)
|
||||
default:
|
||||
}
|
||||
cancel()
|
||||
synctest.Wait()
|
||||
var out toolOutcome
|
||||
select {
|
||||
case out = <-done:
|
||||
if !store.Barrier().Busy() || store.Barrier().HasWriters() {
|
||||
t.Error("cancelled waiter changed the exclusive owner's admission")
|
||||
}
|
||||
store.Barrier().ExitExclusive()
|
||||
default:
|
||||
t.Error("cancelled call still waits for checkpoint admission")
|
||||
store.Barrier().ExitExclusive()
|
||||
out = <-done
|
||||
}
|
||||
if !out.blocked || out.executed || hookCalls != 0 {
|
||||
t.Errorf("cancelled invocation entered user code: outcome=%+v, hooks=%d", out, hookCalls)
|
||||
}
|
||||
if store.Barrier().Busy() {
|
||||
t.Error("cancelled invocation leaked writer admission")
|
||||
}
|
||||
})
|
||||
if data, err := os.ReadFile(path); err != nil || string(data) != "user bytes" {
|
||||
t.Fatalf("cancelled invocation changed bytes: %q %v", data, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelledCheckpointWaitReleasesProvisionalWorkspacePaths(t *testing.T) {
|
||||
root, locks := t.TempDir(), t.TempDir()
|
||||
owner, err := workspacelease.New(root, locks, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
probe, err := workspacelease.New(root, locks, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
owner.BeginRun()
|
||||
defer owner.EndRun()
|
||||
probe.BeginRun()
|
||||
defer probe.EndRun()
|
||||
path := filepath.Join(root, "pending.txt")
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
store := checkpoint.New("", root)
|
||||
if err := store.Barrier().EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writer := &partialLeaseWriter{workspaceLeaseTestTool: workspaceLeaseTestTool{name: "pending_writer"}, path: path}
|
||||
a := deliveryLeaseTestAgent(t, owner, writer)
|
||||
a.SetMutationObserver(checkpoint.NewMutationObserver(checkpoint.ObserverOptions{Store: store}))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan toolOutcome, 1)
|
||||
go func() { done <- a.executeOne(ctx, &a.turn, providerToolCall("pending", writer.Name())) }()
|
||||
synctest.Wait()
|
||||
select {
|
||||
case out := <-done:
|
||||
store.Barrier().ExitExclusive()
|
||||
t.Fatalf("call did not wait at checkpoint admission: %+v", out)
|
||||
default:
|
||||
}
|
||||
cancel()
|
||||
synctest.Wait()
|
||||
select {
|
||||
case out := <-done:
|
||||
if !out.blocked || out.executed {
|
||||
t.Errorf("cancelled invocation = %+v", out)
|
||||
}
|
||||
store.Barrier().ExitExclusive()
|
||||
default:
|
||||
t.Error("cancelled call retained its provisional workspace admission")
|
||||
store.Barrier().ExitExclusive()
|
||||
<-done
|
||||
}
|
||||
if writer.calls.Load() != 0 {
|
||||
t.Error("cancelled writer entered execution")
|
||||
}
|
||||
})
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := probe.AcquirePaths(ctx, []string{path}); err != nil {
|
||||
t.Fatalf("unfinished actor retained aborted call paths: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(path); !os.IsNotExist(err) {
|
||||
t.Fatalf("cancelled call changed the workspace: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -602,7 +602,7 @@ func (a *Agent) prepareToolExecution(ctx context.Context, plan *toolCallPlan) (t
|
||||
// held through post hooks and AfterMutation so rewind cannot interleave with
|
||||
// writer-side user code.
|
||||
if (!plan.readOnly || toolHooksMayMutateWorkspace(a.svc.hooks)) && a.svc.mutationObserver != nil {
|
||||
release, err := a.svc.mutationObserver.BeginWrite()
|
||||
release, err := a.svc.mutationObserver.BeginWriteContext(ctx)
|
||||
if err != nil {
|
||||
return toolOutcome{output: "blocked: " + err.Error(), blocked: true, errMsg: "blocked: checkpoint write preparation failed"}, true
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package checkpoint
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -41,19 +42,12 @@ func (b *MutationBarrier) Generation() uint64 {
|
||||
|
||||
// EnterWrite blocks until exclusive access is free, then increments the writer count.
|
||||
func (b *MutationBarrier) EnterWrite() error {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
for b.exclusive || b.closed {
|
||||
if b.closed {
|
||||
return fmt.Errorf("mutation barrier closed")
|
||||
}
|
||||
b.cond.Wait()
|
||||
}
|
||||
b.writers++
|
||||
return nil
|
||||
return b.EnterWriteContext(context.Background())
|
||||
}
|
||||
|
||||
// EnterWriteContext admits a writer unless its wait is cancelled.
|
||||
func (b *MutationBarrier) EnterWriteContext(ctx context.Context) error {
|
||||
return b.enter(ctx, false)
|
||||
}
|
||||
|
||||
// TryEnterWrite is a non-blocking EnterWrite.
|
||||
@@ -90,19 +84,51 @@ func (b *MutationBarrier) ExitWrite() {
|
||||
|
||||
// EnterExclusive waits until no writers hold the barrier, then takes exclusive.
|
||||
func (b *MutationBarrier) EnterExclusive() error {
|
||||
return b.EnterExclusiveContext(context.Background())
|
||||
}
|
||||
|
||||
// EnterExclusiveContext waits for exclusive access unless ctx is cancelled.
|
||||
func (b *MutationBarrier) EnterExclusiveContext(ctx context.Context) error {
|
||||
return b.enter(ctx, true)
|
||||
}
|
||||
|
||||
func (b *MutationBarrier) enter(ctx context.Context, exclusive bool) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
for b.exclusive || b.writers > 0 || b.closed {
|
||||
// Register under mu so cancellation cannot be lost between the context
|
||||
// check and Wait. Stopping the callback never waits for it under mu.
|
||||
stop := context.AfterFunc(ctx, func() {
|
||||
b.mu.Lock()
|
||||
b.cond.Broadcast()
|
||||
b.mu.Unlock()
|
||||
})
|
||||
defer stop()
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if b.closed {
|
||||
return fmt.Errorf("mutation barrier closed")
|
||||
}
|
||||
if !b.exclusive && (!exclusive || b.writers == 0) {
|
||||
if exclusive {
|
||||
b.exclusive = true
|
||||
} else {
|
||||
b.writers++
|
||||
}
|
||||
return nil
|
||||
}
|
||||
b.cond.Wait()
|
||||
}
|
||||
b.exclusive = true
|
||||
return nil
|
||||
}
|
||||
|
||||
// TryEnterExclusive is a non-blocking EnterExclusive.
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
package checkpoint
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBarrierCancelledWaitPreservesOwner(t *testing.T) {
|
||||
for _, owner := range []string{"writer", "exclusive"} {
|
||||
for _, waiter := range []string{"writer", "exclusive"} {
|
||||
if owner == "writer" && waiter == "writer" {
|
||||
continue
|
||||
}
|
||||
t.Run(owner+"/"+waiter, func(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
b := NewMutationBarrier()
|
||||
release := b.ExitWrite
|
||||
if owner == "writer" {
|
||||
if err := b.EnterWrite(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
} else {
|
||||
if err := b.EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release = b.ExitExclusive
|
||||
}
|
||||
generation := b.Generation()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
if waiter == "writer" {
|
||||
done <- b.EnterWriteContext(ctx)
|
||||
} else {
|
||||
done <- b.EnterExclusiveContext(ctx)
|
||||
}
|
||||
}()
|
||||
synctest.Wait()
|
||||
cancel()
|
||||
synctest.Wait()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("wait error = %v", err)
|
||||
}
|
||||
default:
|
||||
t.Error("cancellation did not wake waiter")
|
||||
release()
|
||||
<-done
|
||||
return
|
||||
}
|
||||
if !b.Busy() || b.Generation() != generation || b.HasWriters() != (owner == "writer") {
|
||||
t.Error("cancelled waiter changed the owner's state")
|
||||
}
|
||||
if b.TryEnterExclusive() {
|
||||
t.Error("cancelled waiter released the owner's admission")
|
||||
b.ExitExclusive()
|
||||
}
|
||||
release()
|
||||
if !b.TryEnterExclusive() {
|
||||
t.Fatal("cancelled waiter left an admission behind")
|
||||
}
|
||||
b.ExitExclusive()
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBarrierContextRefusesCancelledFreeAdmission(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
for _, exclusive := range []bool{false, true} {
|
||||
b := NewMutationBarrier()
|
||||
var err error
|
||||
if exclusive {
|
||||
err = b.EnterExclusiveContext(ctx)
|
||||
} else {
|
||||
err = b.EnterWriteContext(ctx)
|
||||
}
|
||||
if !errors.Is(err, context.Canceled) || b.Busy() || b.Generation() != 0 {
|
||||
t.Fatalf("cancelled free admission: err=%v, busy=%v, generation=%d", err, b.Busy(), b.Generation())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBarrierContextDeadlineAndCloseWakeWaiters(t *testing.T) {
|
||||
for _, exclusive := range []bool{false, true} {
|
||||
for _, closeBarrier := range []bool{false, true} {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
b := NewMutationBarrier()
|
||||
if err := b.EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
if exclusive {
|
||||
done <- b.EnterExclusiveContext(ctx)
|
||||
} else {
|
||||
done <- b.EnterWriteContext(ctx)
|
||||
}
|
||||
}()
|
||||
synctest.Wait()
|
||||
if closeBarrier {
|
||||
b.Close()
|
||||
} else {
|
||||
time.Sleep(time.Hour)
|
||||
}
|
||||
err := <-done
|
||||
if err == nil || (!closeBarrier && !errors.Is(err, context.DeadlineExceeded)) {
|
||||
t.Errorf("wait error = %v", err)
|
||||
}
|
||||
if !b.Busy() || b.HasWriters() || b.Generation() != 0 {
|
||||
t.Error("abandoned wait changed the exclusive owner")
|
||||
}
|
||||
b.ExitExclusive()
|
||||
if closeBarrier && (b.TryEnterWrite() || b.TryEnterExclusive()) {
|
||||
t.Error("closed barrier admitted a successor")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBarrierContextLiveWaitersAndLegacyEntry(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
b := NewMutationBarrier()
|
||||
if err := b.EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 2)
|
||||
go func() { done <- b.EnterWriteContext(context.Background()) }()
|
||||
go func() { done <- b.EnterWrite() }()
|
||||
synctest.Wait()
|
||||
b.ExitExclusive()
|
||||
if first, second := <-done, <-done; first != nil || second != nil {
|
||||
t.Fatalf("live writers failed: %v %v", first, second)
|
||||
}
|
||||
b.ExitWrite()
|
||||
if !b.HasWriters() || b.TryEnterExclusive() {
|
||||
t.Fatal("one writer's release removed another writer")
|
||||
}
|
||||
b.ExitWrite()
|
||||
if err := b.EnterExclusiveContext(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.ExitExclusive()
|
||||
})
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package checkpoint
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
@@ -116,18 +117,34 @@ func (o *MutationObserver) OwnershipTurn() int {
|
||||
|
||||
// BeginWrite retires undo durably before a tool or hook can modify files.
|
||||
func (o *MutationObserver) BeginWrite() (func(), error) {
|
||||
return o.BeginWriteContext(context.Background())
|
||||
}
|
||||
|
||||
// BeginWriteContext checks cancellation before admission and after preparation.
|
||||
// Durable undo retirement already completed is not reversed on cancellation.
|
||||
func (o *MutationObserver) BeginWriteContext(ctx context.Context) (func(), error) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if o == nil || o.store == nil {
|
||||
return nil, nil
|
||||
}
|
||||
barrier := o.store.Barrier()
|
||||
if err := barrier.EnterWrite(); err != nil {
|
||||
if err := barrier.EnterWriteContext(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := o.store.InvalidateUndo(); err != nil {
|
||||
barrier.ExitWrite()
|
||||
return nil, fmt.Errorf("prepare workspace write: %w", err)
|
||||
}
|
||||
return barrier.ExitWrite, nil
|
||||
if err := ctx.Err(); err != nil {
|
||||
barrier.ExitWrite()
|
||||
return nil, err
|
||||
}
|
||||
return sync.OnceFunc(barrier.ExitWrite), nil
|
||||
}
|
||||
|
||||
// RegisterWriter marks a background writer as active. Rollback precheck returns
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
package checkpoint
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
)
|
||||
|
||||
func TestCancelledObserverAdmissionKeepsDurableUndo(t *testing.T) {
|
||||
s, dir, root, _, previous := committedCodeUndo(t)
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
if err := s.Barrier().EnterExclusive(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
observer := NewMutationObserver(ObserverOptions{Store: s})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
release, err := observer.BeginWriteContext(ctx)
|
||||
if release != nil {
|
||||
release()
|
||||
}
|
||||
done <- err
|
||||
}()
|
||||
synctest.Wait()
|
||||
cancel()
|
||||
synctest.Wait()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("preparation error = %v", err)
|
||||
}
|
||||
default:
|
||||
t.Error("observer did not leave cancelled wait")
|
||||
s.Barrier().ExitExclusive()
|
||||
<-done
|
||||
return
|
||||
}
|
||||
if s.Barrier().HasWriters() || !s.Barrier().Busy() {
|
||||
t.Error("observer changed the exclusive owner's state")
|
||||
}
|
||||
s.Barrier().ExitExclusive()
|
||||
})
|
||||
requireUndoOffer(t, s, previous.TransactionID)
|
||||
requireUndoOffer(t, New(dir, root), previous.TransactionID)
|
||||
}
|
||||
|
||||
func TestObserverCancellationAfterDurablePreparationDoesNotReviveUndo(t *testing.T) {
|
||||
s, dir, root, _, _ := committedCodeUndo(t)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s.OnUndoChange(cancel)
|
||||
release, err := NewMutationObserver(ObserverOptions{Store: s}).BeginWriteContext(ctx)
|
||||
if release != nil {
|
||||
release()
|
||||
t.Error("cancelled preparation returned a writer permit")
|
||||
}
|
||||
if !errors.Is(err, context.Canceled) || s.Barrier().Busy() {
|
||||
t.Fatalf("post-preparation cancellation: err=%v, busy=%v", err, s.Barrier().Busy())
|
||||
}
|
||||
if undo, err := New(dir, root).AvailableUndo(); err != nil || undo != nil {
|
||||
t.Fatalf("cancelled preparation revived retired undo: %+v %v", undo, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestObserverReleaseDoesNotReleaseAnotherWriter(t *testing.T) {
|
||||
s := New("", t.TempDir())
|
||||
o := NewMutationObserver(ObserverOptions{Store: s})
|
||||
first, err := o.BeginWrite()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
second, err := o.BeginWrite()
|
||||
if err != nil {
|
||||
first()
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer second()
|
||||
first()
|
||||
generation := s.Barrier().Generation()
|
||||
first()
|
||||
if !s.Barrier().HasWriters() || s.Barrier().Generation() != generation {
|
||||
t.Fatal("repeated cleanup released another writer")
|
||||
}
|
||||
if s.Barrier().TryEnterExclusive() {
|
||||
s.Barrier().ExitExclusive()
|
||||
t.Fatal("rewind admitted while second writer was active")
|
||||
}
|
||||
second()
|
||||
if s.Barrier().Busy() {
|
||||
t.Fatal("final writer cleanup leaked admission")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
## 修复
|
||||
|
||||
- Cancelled calls leave checkpoint waits before hooks/tools. Repeated permit
|
||||
cleanup preserves other writers; background registration stays separate. Refs #14025
|
||||
- 取消工具调用后可退出检查点准入等待,不再等待独占方释放后继续进入同步 hook 或
|
||||
工具;重复清理同一许可不会释放其他写入者。后台登记及异步 hook 生命周期另行处理。Refs #14025
|
||||
Reference in new issue
Block a user