diff --git a/docs/CONCURRENT_MUTATIONS_REVIEW_SERIES.md b/docs/CONCURRENT_MUTATIONS_REVIEW_SERIES.md new file mode 100644 index 0000000000..d4edf90dfe --- /dev/null +++ b/docs/CONCURRENT_MUTATIONS_REVIEW_SERIES.md @@ -0,0 +1,71 @@ +--- +owner: @SivanCola +backup: @esengine +status: active +reviewed: 2026-10-24 +--- + +# Concurrent mutation review series / 并发写入分批审查 + +## Purpose / 目的 + +The complete implementation remains pinned at +`467a61668a0781cab171536158536283d931e4d9` in #14133 as a source reference. +It is being split following the maintainer's review-size request. +The source PR's validation is historical evidence, not qualification of a slice. + +完整实现保留在 #14133 的上述提交。本系列按维护者要求拆分；原实现的测试记录 +保留，但每一批都必须独立构建、检查及测试，不能继承整体分支的通过结论。 + +## Sequence and review boundaries / 顺序与边界 + +| Order / 顺序 | Review unit / 审查单元 | Prerequisite and activation / 前置条件及接入 | +| --- | --- | --- | +| 1 | Mutation admission primitive, content versions, legacy claim compatibility | Current slice; library only. No tool/controller/settings activation. | +| 2 | Tool effect descriptions and per-actor read-version provenance | Depends on 1. Review path grants, approval ordering, overlays and same-file stale-view regressions. | +| 3 | Shared tool/agent transaction assembly | Depends on 2. Review parent/child/fleet/dynamic paths, refused preparation and physical completion. | +| 4 | Source snapshots and private worktree preparation | Independent platform layer where possible. Review Git common-directory operations, modes, symlinks, untracked and ignored inputs. | +| 5 | Journaled per-file application and conditional recovery | Depends on 1 and 4. Review manifest containment, preimages, identity, failure injection, conditional rollback and idempotence. | +| 6 | Frozen check inputs and verification receipts | Review input observation, environment identity, stale evidence and unavailable backend behavior before integration activation. | +| 7 | Durable isolated task records and lifecycle | Depends on 4 and 6. Split preparation/execution from recovery/retention if either diff becomes too broad. | +| 8 | Task/fleet/background and nested ownership | Depends on 3 and 7. Review budgets, continuation, direct-parent identity, cancellation, restart and cross-version fences. | +| 9 | Root conversations and source integration | Depends on 5–8. Review current grants, source drift, controlled checks and recovery blocking together. | +| 10 | Controlled portable verifier and confined shell/hooks | Separate written command/authority security review; no unsupported-platform enforcement claim. | +| 11 | Controller/wire/frontends and migration/default selection | Activate only after required lower-layer review and platform evidence; preserve legacy settings and active runtime protocols. | +| 12 | Product/process performance and final acceptance | Reconcile every source delta, tests and compatibility evidence on the combined accepted tree. | + +表中是依赖关系及审查边界，不是承诺将每行所有文件一次性提交。若一行仍然 +包含多个独立机制，应继续拆分。前置接口经审查确定后，再提交对应的接入批次。 +优先使用直接面向 `studio` 的独立 PR；需要堆叠时明确实际基分支及依赖。 + +## Extraction and security rules / 拆分与安全规则 + +1. Each slice MUST build and run its own tests without unsubmitted sibling + patches. No lint baseline, safety gate or coverage floor is relaxed to make + an extraction pass. +2. Every PR MUST state Cause, Blast radius, Neighbouring behaviours tested and + Why this layer, plus its actual base/head and local/hosted evidence. +3. Sensitive slices MUST document input provenance, authority, filesystem or + command sinks, validation, failure recovery and explicit unenforced bounds. +4. Original CodeQL findings remain unresolved until assessed at their actual + sources and sinks. Moving code between PRs is not security remediation. +5. The source branch MUST remain recoverable. Each extracted file or hunk is + accounted for as kept, adapted, deferred or intentionally dropped; the + combined final tree is compared with the pinned source plus upstream changes. +6. Primitive API simplifications in slice 1 are intentional: use `Options`, + lease methods and explicit-context version reads. Runtime-only context + conveniences and legacy recovery admission hooks are deferred to their users. + +每批必须独立验证，不降低门禁；安全说明不能用“旧分支测过”替代。拆分时保留 +完整来源，并记录适配或延期的差异；最后对合并后的代码树做完整性核对。第一批 +去掉尚无调用者的便利函数，后续接入时按已审查接口适配，不复制旧 API。 + +## First-slice review / 第一批审查入口 + +Read [the coordinator security review](MUTATION_COORDINATOR_SECURITY.md), then +`internal/state/mutation/doc.go`, `coordinator.go`, `registry.go`, `resources.go`, +`capacity.go` and `completion.go`. The compatibility change is limited to +`workspacelease/claim_attempt.go` and its existing `paths.go` publication owner. + +先阅读安全审查及包说明，再审查资源申请、登记、身份、容量及释放。旧协议的 +调整限于独立短申请及其发布清理，不接入运行时恢复屏障或修改旧锁语义。 diff --git a/docs/MUTATION_COORDINATOR_SECURITY.md b/docs/MUTATION_COORDINATOR_SECURITY.md new file mode 100644 index 0000000000..70aa7daa42 --- /dev/null +++ b/docs/MUTATION_COORDINATOR_SECURITY.md @@ -0,0 +1,160 @@ +--- +owner: @SivanCola +backup: @esengine +status: active +reviewed: 2026-10-24 +--- + +# Mutation coordinator security review / 写入事务协调器安全审查 + +## Scope and decision / 范围与结论 + +This is the author's security analysis of the first review slice of #14133, +extracted from source commit `467a61668a0781cab171536158536283d931e4d9`. +It is not an independent approval or a disposition of that PR's CodeQL findings. + +The slice adds `internal/state/mutation`, content-version reads, and the minimum +non-blocking `workspacelease` compatibility claim API. It introduces no production +callers, shell launcher, hooks, worktree integration, settings or default change. + +这是 #14133 第一批改动的作者安全分析，不代表独立审批，也不代表原 PR 的 +CodeQL 告警已解决。本批只增加事务基础库、内容版本读取和旧协议兼容申请接口； +没有接入生产调用者、命令执行、hooks、worktree 整合或设置，默认行为保持原样。 + +## Threat model and authority / 威胁模型与授权 + +The protected property is exclusion among cooperating Reasonix processes that +share one trusted state directory. A caller may request conflicting paths, +cancel, fail, panic in a callback, retain a stale input version, or die while +queued/active. + +The coordinator must not grant partial sets, steal a live lease, +release a successor, silently accept stale versions, or erase unreadable records. + +`New(root, stateDir, ...)` is a host API. Both arguments and admission callbacks +must come from the authorized host, not directly from model-controlled JSON. + +`Intent.Paths` may contain authorized paths outside `root`: root is a resolution +base, not an access-control boundary. Canonicalization establishes identity, +**not permission**. The host must resolve the actual effect and complete grants +before acquisition. + +`Options.AdmissionCheck` can reject admission under the +lease; omitting it does not create an implicit permission check. + +保护对象是共享可信状态目录、遵守协议的 Reasonix 进程。请求可能冲突、取消、 +过期或崩溃；不得部分占用、夺取有效租约、释放后继事务或删除无法理解的记录。 + +构造参数和回调必须来自已授权宿主。资源可以是工作区之外已获授权的路径： +路径规范化只确认身份，不授予读写权限。权限检查及完整影响范围仍由宿主负责。 + +Not protected: malicious modification of the state directory by the same OS +user, non-cooperating filesystem writers, arbitrary symlink/hardlink replacement +after validation, filesystem/OS failure, databases, network services, credentials +or ports. Advisory locks are not a sandbox. + +Different state directories do not +coordinate even if paths happen to name the same file. Never remove lock files +to resolve contention. The state directory must not be exposed as an execution +workspace or be placed in an untrusted/shared writable directory. + +Existing named-file hardlinks are checked with `os.SameFile`. Directory resource +sets do not enumerate aliases outside their trees. A directory lease therefore +cannot establish complete alias coverage; future opaque-write callers must +constrain such aliases or add conservative coordination before claiming it. + +不覆盖同一系统用户恶意篡改状态目录、不遵守协议的写入者、校验后的任意路径 +替换、系统故障及外部服务。不同状态目录不互相协调；不能删除锁文件制造并发。 +状态目录不得作为执行工作区，也不得放在不可信的共享可写目录中。 + +## Data flows and checks / 数据流及检查 + +| Input / 输入 | Sink and guard / 访问位置及约束 | +| --- | --- | +| Host `stateDir` | `New` creates `mutations-v1` with mode 0700; records use 0600 and atomic replacement. Existing directory ownership/ACLs are a host precondition, not revalidated here. Windows modes are not an ACL sandbox. | +| Host root and intent paths | `resources` uses existing `CanonicalWritePath`, resolves existing ancestors, refuses ambiguous leaf links/Windows spellings, sorts and deduplicates. The locked validation repeats resolution and compares the complete set. It intentionally permits authorized extra roots. | +| Expected-content map | Keys must be absolute and covered by held resources. `heldWithin` accepts case-folded coverage only when the filesystem confirms the same existing prefix; conflict detection may conservatively over-serialize. `validateExpected` fully hashes inputs again after grant, regardless of the metadata-only waiting optimization. | +| Content read path | `ReadContentVersion` is a low-level arbitrary-path read API, not a permission boundary. It hashes regular bytes with cancellation, identifies absent/directory/link objects and rejects unsupported kinds. File identity/size/mtime changes during reads refuse. Link versions hash link text; mutation path resolution refuses ambiguous leaf links. A directory version is not a recursive snapshot. | +| Transaction filenames | IDs originate from `crypto/rand.Text`, not session/execution/call IDs. Registry filenames come from directory entries, whose path components cannot contain separators. A loaded record must match its filename, supported resource schema and nonzero queue order before any stale-owner removal. Diagnostic identities are serialized as JSON, never evaluated as commands. | +| Capacity identity | Opaque bounded strings, positive units/limits, no duplicates; inconsistent limits refuse. IDs do not become filesystem paths. Admission is atomic with paths; path-blocked requests do not consume execution slots. | +| Legacy claims | `TryClaimPaths` rejects empty/unresolvable sets, uses the existing workspace and claim locks, publishes all paths as one claim and returns an idempotent release. Probe requests publish nothing. Failure/panic before handoff removes only the newly created claim. | +| Continuation | Copies resource identities into a fresh acquisition; it does not transfer a live lease or reuse expected versions/approval. The future caller must re-establish its authorization and content preconditions. | + +上表逐项区分宿主路径、预期版本、随机事务身份及容量身份。文件访问 API 并不 +自行形成权限边界。等待期间的元数据优化不能替代获得租约后的完整内容校验。 +目录版本不是递归快照，续跑对象也不携带旧授权或旧审批。 + +## Ordering, liveness and failure / 顺序、存活及失败 + +The registry guard covers queue inspection and publication, not tool execution. +Compatibility discovery runs outside it and is followed by a locked recheck. + +Earlier overlapping requests retain priority; disjoint resources may proceed. +Read-only/external effects cannot smuggle paths or expected versions through an +empty set. Capacity-only leases cannot authorize file mutations. + +A unique live OS lock, not a PID or elapsed timeout, establishes transaction +liveness. A wait timeout abandons the waiter only. Process death releases OS +ownership; a later registry pass removes a structurally valid dead record. + +Unknown/corrupt schemas refuse and retain the bytes, even if apparently stale. +No migration or repair is attempted. + +Nested leases cannot widen their parent's +resources or end it. Duplicate `End` calls cannot release a successor's random +transaction identity. + +`HoldUntil` delays release until all registered physical +execution completions close; a caller that forgets completion can block progress. + +登记临界区只保护排队及发布。旧协议检查在临界区之外运行，随后再次校验。 +超时仅终止等待；真实存活由 OS 锁判断。未知版本及损坏记录保留并拒绝继续。 +嵌套调用不能扩大范围或释放父租约；重复结束不能释放后继事务。实际执行尚未 +结束时必须登记完成信号，否则不能声称取消已安全释放写入权。 + +This layer does not perform writes on the caller's behalf, maintain durable +write preimages, or recover partially changed files. `Outcome` is the caller's +reported classification, not proof of rollback, commit or verification. A future +runtime/integration PR must establish those additional contracts explicitly. + +本层不代写文件，不保存写前映像，也不恢复部分变更。结果枚举是调用者报告， +不是提交、回滚或验证成功的证明；这些保证属于后续运行时及整合层。 + +## Compatibility and security gates / 兼容及安全门禁 + +| Contract / 协议 | Existing behavior / 旧行为 | This slice / 本批行为 | +| --- | --- | --- | +| Legacy workspace claims | Existing long-lived owners and optimistic bypasses | Format unchanged; independent short claims interoperate when callers opt in. An old bypass cannot be repaired by this library alone. | +| Mutation record v1 | Absent on older studio | Path-only records; live/dead checked using OS lock. | +| Mutation record v2 | Unsupported by v1-only readers | Carries capacities; unsupported versions fail closed and retain records. No active mixed-version safety claim beyond these readers. | +| Settings, transcripts, results and wire | Existing schemas | Untouched; no new defaults or automatic migration. | + +Required review includes actual first-slice CodeQL results; inherited #14133 +alerts are not blanket false positives. This slice has no production command +execution sink. The two old command-sink discussions remain obligations of the +later shell/hook PRs. + +Before runtime activation, independently review all effect +resolvers, access checks, snapshot/version provenance and physical completion +callers; passing this library's tests is insufficient for that activation. + +必须核对本批实际 CodeQL 结果，不能把原 PR 的告警批量认定为误报。命令及 hook +执行的数据流由后续对应 PR 单独审查。接入运行时之前，还必须审查影响解析、 +权限、版本来源和实际执行完成回调；基础库测试通过不代表上层已安全接入。 + +## Verification map / 验证对应表 + +- Complete-set cancellation and no widening: `coordinator_test.go`, + `capacity_scope_test.go`, `try_test.go`. +- Conflict fairness and independent progress: `fairness_test.go`, + `capacity_test.go`, `compatibility_phase_test.go`. +- Stale reads, aliases and file identity: `stale_wait_test.go`, + `case_fold_test.go`, `workspace_resources_test.go`, `content_version_test.go`. +- Refusal/panic and completion ownership: `callback_failure_test.go`, + `completion_test.go`, `workspacelease/claim_panic_test.go`. +- Real subprocess death and protocol preservation: `process_test.go`, + `capacity_process_test.go`, `compatibility_test.go`. + +上述测试分别验证完整资源申请、公平排队、旧视图拒绝、别名身份、异常清理、 +实际执行完成、真实进程退出及未知协议保留。平台测试和 CI 结果在每个 PR 中 +按实际来源报告，不继承完整实现的旧验收结论。 diff --git a/docs/design/FIRST_RUN_WORKSPACE.md b/docs/design/FIRST_RUN_WORKSPACE.md index 5a616a5d9d..432a4b7d3a 100644 --- a/docs/design/FIRST_RUN_WORKSPACE.md +++ b/docs/design/FIRST_RUN_WORKSPACE.md @@ -1,13 +1,18 @@ --- -owner: @Oliver-Rui -status: proposed +owner: @esengine +backup: @SivanCola +author: @Oliver-Rui +status: active +reviewed: 2026-10-24 --- # First run without a picked workspace Status: proposed. Nothing in this document is shipped unless §8 says so. It lives under `docs/design/` so the product-doc retrieval corpus (`docs/*.md`) cannot present it to a model as a feature. -Reference: how MiMo Desktop (observed 2026-10-24 on a machine running it) handles the same three first-run decisions — connect a model, pick a workspace, start a session. Its behaviour below is read off installed artifacts (config schema, directory layout, state store), not source; the mapping onto this repository in §6 is grounded in source. +Reference: how MiMo Desktop (observed 2026-10-24 on a machine running it) handles the same three first-run decisions — connect a model, pick a workspace, start a session. + +Its behaviour below is read off installed artifacts (config schema, directory layout, state store), not source; the mapping onto this repository in §6 is grounded in source. ## 1. The problem @@ -19,9 +24,13 @@ Studio's first launch asks three things at once, and only one of them is honestl | Start a session | the composer is reachable as soon as the workbench opens | yes | | Pick a workspace | `pickLaunchWorkspace(cwd, appExe, remembered, home)` — with nothing remembered, the first pane's root is the **user's home directory** | no | -The fallback chain in `cmd/reasonix-studio-host/launchroot.go` is careful about *which* cwd it trusts — a Start-menu launch inherits the install directory, and taking that as a project "puts the app's own folder in the sidebar again after every removal" — but its last resort is the home directory. That folder was never chosen as a project. It is broad, it is sensitive, and it is not represented in the sidebar as anything the user picked, so a session running there reads as a temporary session: not in the user's project, scoped to a folder they did not choose, and removing it does not stick. +The fallback chain in `cmd/reasonix-studio-host/launchroot.go` is careful about *which* cwd it trusts — a Start-menu launch inherits the install directory, and taking that as a project "puts the app's own folder in the sidebar again after every removal" — but its last resort is the home directory. -Everything else is already in place for a decoupled first run. Session transcripts persist in the global store (`SessionDir()` = `/sessions`), so a session without a chosen project is not lost. Each logical session already owns a private scratch directory (`SessionTemp`, rotated per session, deleted when it ends, surfaced to the model as `$TMPDIR`). Provider setup already refuses to block. The workspace is the last coupled decision. +That folder was never chosen as a project. It is broad, it is sensitive, and it is not represented in the sidebar as anything the user picked, so a session running there reads as a temporary session: not in the user's project, scoped to a folder they did not choose, and removing it does not stick. + +Everything else is already in place for a decoupled first run. Session transcripts persist in the global store (`SessionDir()` = `/sessions`), so a session without a chosen project is not lost. + +Each logical session already owns a private scratch directory (`SessionTemp`, rotated per session, deleted when it ends, surfaced to the model as `$TMPDIR`). Provider setup already refuses to block. The workspace is the last coupled decision. ## 2. What MiMo does instead diff --git a/internal/base/fileutil/content_version.go b/internal/base/fileutil/content_version.go new file mode 100644 index 0000000000..8ccfc2bdf7 --- /dev/null +++ b/internal/base/fileutil/content_version.go @@ -0,0 +1,77 @@ +package fileutil + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "os" +) + +type ContentVersion struct { + Kind string `json:"kind"` + Digest string `json:"digest,omitempty"` + Mode uint32 `json:"mode,omitempty"` + BufferVersion string `json:"bufferVersion,omitempty"` +} + +var ErrContentChanged = errors.New("file changed while reading content version") + +func ReadContentVersion(ctx context.Context, path string) (ContentVersion, error) { + if err := ctx.Err(); err != nil { + return ContentVersion{}, err + } + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + return ContentVersion{Kind: "absent"}, nil + } + if err != nil { + return ContentVersion{}, err + } + v := ContentVersion{Mode: uint32(info.Mode().Perm())} + switch { + case info.Mode()&os.ModeSymlink != 0: + target, err := os.Readlink(path) + if err != nil { + return ContentVersion{}, err + } + sum := sha256.Sum256([]byte(target)) + v.Kind, v.Digest = "symlink", hex.EncodeToString(sum[:]) + case info.Mode().IsRegular(): + f, err := os.Open(path) + if err != nil { + return ContentVersion{}, err + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, &versionReader{ctx: ctx, r: f}); err != nil { + return ContentVersion{}, err + } + after, err := f.Stat() + if err != nil { + return ContentVersion{}, err + } + if !os.SameFile(info, after) || info.Size() != after.Size() || !info.ModTime().Equal(after.ModTime()) { + return ContentVersion{}, ErrContentChanged + } + v.Kind, v.Digest = "file", hex.EncodeToString(h.Sum(nil)) + case info.IsDir(): + v.Kind = "directory" + default: + return ContentVersion{}, errors.ErrUnsupported + } + return v, nil +} + +type versionReader struct { + ctx context.Context + r io.Reader +} + +func (r *versionReader) Read(p []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.r.Read(p) +} diff --git a/internal/base/fileutil/content_version_test.go b/internal/base/fileutil/content_version_test.go new file mode 100644 index 0000000000..402412e3c0 --- /dev/null +++ b/internal/base/fileutil/content_version_test.go @@ -0,0 +1,70 @@ +package fileutil + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestContentVersionTracksBytesKindAndAbsence(t *testing.T) { + path := filepath.Join(t.TempDir(), "file") + absent, err := ReadContentVersion(t.Context(), path) + if err != nil || absent.Kind != "absent" { + t.Fatalf("absent: %+v %v", absent, err) + } + var versions []ContentVersion + for _, body := range []string{"before", "after!", "before"} { + if err := os.WriteFile(path, []byte(body), 0600); err != nil { + t.Fatal(err) + } + v, err := ReadContentVersion(t.Context(), path) + if err != nil || v.Kind != "file" || v.Digest == "" { + t.Fatalf("file: %+v %v", v, err) + } + versions = append(versions, v) + } + if versions[0] == versions[1] || versions[0] != versions[2] { + t.Fatalf("content identity depends on metadata or misses same-size changes: %+v", versions) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(path, 0700); err != nil { + t.Fatal(err) + } + dir, err := ReadContentVersion(t.Context(), path) + if err != nil || dir.Kind != "directory" || dir.Digest != "" { + t.Fatalf("directory: %+v %v", dir, err) + } +} + +func TestContentVersionCancellationPrecedesFilesystemAccess(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := ReadContentVersion(ctx, filepath.Join(t.TempDir(), "missing")); !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled read: %v", err) + } +} + +func TestContentVersionHashesSymlinkTextWithoutReadingTarget(t *testing.T) { + path := filepath.Join(t.TempDir(), "link") + if err := os.Symlink("missing-target", path); err != nil { + t.Skipf("symlinks unavailable: %v", err) + } + first, err := ReadContentVersion(t.Context(), path) + if err != nil || first.Kind != "symlink" || first.Digest == "" { + t.Fatalf("link: %+v %v", first, err) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := os.Symlink("another-missing-target", path); err != nil { + t.Fatal(err) + } + second, err := ReadContentVersion(t.Context(), path) + if err != nil || second == first { + t.Fatalf("retargeted link: %+v %v", second, err) + } +} diff --git a/internal/state/mutation/callback_failure_test.go b/internal/state/mutation/callback_failure_test.go new file mode 100644 index 0000000000..436ea6fad2 --- /dev/null +++ b/internal/state/mutation/callback_failure_test.go @@ -0,0 +1,123 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestAdmissionPanicReleasesOnlyItsOwnTransaction(t *testing.T) { + root := t.TempDir() + c, err := New(root, t.TempDir(), nil, Options{AdmissionCheck: func(_ context.Context, intent Intent) error { + if intent.CallID == "panic" { + panic("fixture admission failure") + } + return nil + }}) + if err != nil { + t.Fatal(err) + } + held, err := c.Acquire(t.Context(), Intent{Effect: NamedPaths, Paths: []string{filepath.Join(root, "held")}}) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + for index, ctx := range []context.Context{t.Context(), held.Context(t.Context())} { + paths := []string{filepath.Join(root, "held")} + if index == 0 { + paths = []string{filepath.Join(root, "other")} + } + if lease, err := c.Acquire(ctx, Intent{Effect: NamedPaths, Paths: paths, CallID: "panic"}); lease != nil || !errors.Is(err, ErrCallback) { + t.Fatalf("admission panic did not fail closed: %v %v", lease, err) + } + } + if err := held.CheckPaths([]string{filepath.Join(root, "held")}); err != nil { + t.Fatalf("failed child released an existing transaction: %v", err) + } + probe, err := c.Acquire(t.Context(), Intent{Effect: NamedPaths, Paths: []string{filepath.Join(root, "other")}}) + if err != nil { + t.Fatalf("failed admission left resources: %v", err) + } + probe.End(NotExecuted) +} + +func TestNoticePanicClosesWaitAndReleasesTransaction(t *testing.T) { + for _, phase := range []string{"waiting", "acquired", "released"} { + t.Run(phase, func(t *testing.T) { + root := t.TempDir() + waiting := make(chan struct{}) + var waitingOnce sync.Once + var failed atomic.Bool + var mu sync.Mutex + var states []string + c, err := New(root, t.TempDir(), func(notice Notice) { + if notice.CallID != "target" { + return + } + mu.Lock() + states = append(states, notice.State) + mu.Unlock() + if notice.State == "waiting" { + waitingOnce.Do(func() { close(waiting) }) + } + if notice.State == phase && failed.CompareAndSwap(false, true) { + panic("fixture notice failure") + } + }) + if err != nil { + t.Fatal(err) + } + intent := Intent{Effect: NamedPaths, Paths: []string{filepath.Join(root, "shared")}} + held, err := c.Acquire(t.Context(), intent) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + type answer struct { + lease *Lease + err error + } + done := make(chan answer, 1) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + go func() { intent.CallID = "target"; lease, err := c.Acquire(ctx, intent); done <- answer{lease, err} }() + select { + case <-waiting: + case <-ctx.Done(): + t.Fatal("wait did not open") + } + held.End(NotExecuted) + result := <-done + if phase == "released" { + if result.err != nil || result.lease == nil { + t.Fatalf("acquire failed: %+v", result) + } + result.lease.End(Committed) + } else if result.lease != nil || !errors.Is(result.err, ErrCallback) { + t.Fatalf("notice failure reached execution: %+v", result) + } + if !failed.Load() { + t.Fatal("fixture did not raise the selected callback failure") + } + mu.Lock() + closed := false + for _, state := range states { + closed = closed || state == "abandoned" || (phase == "released" && state == "acquired") + } + mu.Unlock() + if !closed { + t.Fatal("opened wait never closed") + } + intent.CallID = "probe" + probe, err := c.Acquire(ctx, intent) + if err != nil { + t.Fatalf("callback left live resources: %v", err) + } + probe.End(NotExecuted) + }) + } +} diff --git a/internal/state/mutation/capacity.go b/internal/state/mutation/capacity.go new file mode 100644 index 0000000000..0ee24798a2 --- /dev/null +++ b/internal/state/mutation/capacity.go @@ -0,0 +1,163 @@ +package mutation + +import ( + "cmp" + "context" + "errors" + "path/filepath" + "slices" + "strings" + + "reasonix/internal/base/filelock" + "reasonix/internal/state/workspacelease" +) + +// Capacity is a host-owned execution pool requested atomically with paths. +// A pool ID denotes one immutable limit; changing the limit needs a new ID. +type Capacity struct { + ID string `json:"id"` + Limit uint32 `json:"limit"` + Units uint32 `json:"units"` +} + +func normalizedCapacities(in []Capacity) ([]Capacity, error) { + out := slices.Clone(in) + slices.SortFunc(out, func(a, b Capacity) int { return cmp.Compare(a.ID, b.ID) }) + for i, c := range out { + if strings.TrimSpace(c.ID) == "" || len(c.ID) > 256 || c.Limit == 0 || c.Units == 0 || c.Units > c.Limit || (i > 0 && out[i-1].ID == c.ID) { + return nil, ErrInvalidIntent + } + } + return out, nil +} + +func coversCapacities(held, requested []Capacity) bool { + for _, want := range requested { + if !slices.ContainsFunc(held, func(h Capacity) bool { return h.ID == want.ID && h.Limit == want.Limit && h.Units >= want.Units }) { + return false + } + } + return true +} + +func validRecordResources(r record) bool { + if r.Version == 1 { + return len(r.Paths) > 0 && len(r.Capacities) == 0 && r.Effect != CapacityOnly + } + if r.Version != 2 || len(r.Capacities) == 0 || !filepath.IsAbs(r.Workspace) { + return false + } + if _, err := normalizedCapacities(r.Capacities); err != nil { + return false + } + if r.Effect == CapacityOnly { + return len(r.Paths) == 0 + } + return len(r.Paths) > 0 && (r.Effect == NamedPaths || r.Effect == Workspace || r.Effect == GitMetadata) +} + +func acquisitionContext(ctx context.Context, effect Effect) (context.Context, context.CancelFunc) { + if effect == CapacityOnly { + return context.WithCancel(ctx) + } + return context.WithTimeout(ctx, DefaultWaitLimit) +} + +func capacityLimits(records []record) (map[string]uint64, error) { + limits := map[string]uint64{} + for _, r := range records { + for _, c := range r.Capacities { + if existing, ok := limits[c.ID]; ok && existing != uint64(c.Limit) { + return nil, ErrProtocol + } + limits[c.ID] = uint64(c.Limit) + } + } + return limits, nil +} + +func consumeCapacity(remaining map[string]uint64, capacities []Capacity) bool { + for _, c := range capacities { + if remaining[c.ID] < uint64(c.Units) { + return false + } + } + for _, c := range capacities { + remaining[c.ID] -= uint64(c.Units) + } + return true +} + +func (c *Coordinator) capacityBlocker(r record, records []record) (*record, error) { + remaining, err := capacityLimits(records) + if err != nil { + return nil, err + } + if len(r.Capacities) == 0 { + return nil, nil + } + var claims []record + for _, other := range records { + if other.ID != r.ID && other.Active { + if !consumeCapacity(remaining, other.Capacities) { + return nil, ErrProtocol + } + claims = append(claims, other) + } + } + ordered := slices.Clone(records) + slices.SortFunc(ordered, func(a, b record) int { return cmp.Compare(a.Order, b.Order) }) + for _, other := range ordered { + if other.Active || other.Order >= r.Order || pathBlocker(other, records) != nil { + continue + } + if len(other.Capacities) == 0 { + continue + } + if len(other.Paths) != 0 { + blocked, err := c.compatibilityBlocked(other) + if blocked { + continue + } + if err != nil { + return nil, err + } + } + // Reserve queue priority even when the older request cannot yet fit. + // Otherwise smaller requests can indefinitely consume its next free unit. + for _, capacity := range other.Capacities { + remaining[capacity.ID] -= min(remaining[capacity.ID], uint64(capacity.Units)) + } + claims = append(claims, other) + } + for _, requested := range r.Capacities { + if remaining[requested.ID] >= uint64(requested.Units) { + continue + } + for _, claim := range claims { + if slices.ContainsFunc(claim.Capacities, func(c Capacity) bool { return c.ID == requested.ID }) { + return &claim, nil + } + } + return nil, ErrProtocol + } + return nil, nil +} + +// A preparing transaction may already own its compatibility claim. The OS +// marker distinguishes that reservation from an unrelated legacy blocker. +func (c *Coordinator) compatibilityBlocked(r record) (bool, error) { + release, err := filelock.TryAcquire(c.preparationPath(r.ID)) + if errors.Is(err, filelock.ErrHeld) { + return false, nil + } + if err != nil { + return false, err + } + release() + _, err = workspacelease.ProbeClaimPaths(r.Workspace, c.compatDir, r.Paths) + if errors.Is(err, workspacelease.ErrClaimBusy) { + return true, nil + } + return false, err +} diff --git a/internal/state/mutation/capacity_process_test.go b/internal/state/mutation/capacity_process_test.go new file mode 100644 index 0000000000..1f0447e471 --- /dev/null +++ b/internal/state/mutation/capacity_process_test.go @@ -0,0 +1,41 @@ +package mutation + +import ( + "context" + "errors" + "os" + "os/exec" + "testing" + "time" +) + +func TestCapacityProcessDeathHelper(t *testing.T) { + root := os.Getenv("REASONIX_CAPACITY_DEATH_ROOT") + if root == "" { + return + } + c := testCoordinator(t, root, os.Getenv("REASONIX_CAPACITY_DEATH_STATE")) + if _, err := c.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("root-total")}}); err != nil { + t.Fatal(err) + } + os.Exit(77) +} + +func TestCapacityReclaimedAfterRealProcessExit(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestCapacityProcessDeathHelper$") + cmd.Env = append(os.Environ(), "REASONIX_CAPACITY_DEATH_ROOT="+root, "REASONIX_CAPACITY_DEATH_STATE="+state) + out, err := cmd.CombinedOutput() + var exited *exec.ExitError + if !errors.As(err, &exited) || exited.ExitCode() != 77 { + t.Fatalf("child: %v\n%s", err, out) + } + c := testCoordinator(t, root, state) + l, err := c.Acquire(ctx, Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("root-total")}}) + if err != nil { + t.Fatalf("dead process retained capacity: %v", err) + } + l.End(NotExecuted) +} diff --git a/internal/state/mutation/capacity_scope_test.go b/internal/state/mutation/capacity_scope_test.go new file mode 100644 index 0000000000..6ad4bffbe6 --- /dev/null +++ b/internal/state/mutation/capacity_scope_test.go @@ -0,0 +1,56 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" +) + +func TestCapacityLeaseCannotAuthorizeFilesystemWrites(t *testing.T) { + c := testCoordinator(t, t.TempDir(), t.TempDir()) + l, err := c.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if err != nil { + t.Fatal(err) + } + defer l.End(NotExecuted) + ctx := l.Context(t.Context()) + if l.HoldingPaths() { + t.Fatal("capacity-only admission impersonated a file mutation lease") + } + if _, err := c.Acquire(ctx, intent(filepath.Join(c.root, "a"))); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("file upgrade allowed: %v", err) + } + if _, err := c.Acquire(ctx, Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("other")}}); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("capacity upgrade allowed: %v", err) + } + child, err := c.Acquire(ctx, Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if err != nil { + t.Fatal(err) + } + child.End(NotExecuted) + blocked, cancel := context.WithTimeout(t.Context(), 40*time.Millisecond) + defer cancel() + _, err = c.Acquire(blocked, Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if !errors.Is(err, ErrCapacityWaitTimeout) || errors.Is(err, ErrWaitTimeout) { + t.Fatalf("incorrect capacity timeout: %v", err) + } +} + +func TestCapacityRejectsAmbiguousOrInvalidRequests(t *testing.T) { + c := testCoordinator(t, t.TempDir(), t.TempDir()) + for _, request := range []Intent{ + {Effect: CapacityOnly}, + {Effect: CapacityOnly, Paths: []string{"a"}, Capacities: []Capacity{capacity("total")}}, + {Effect: ReadOnly, Capacities: []Capacity{capacity("total")}}, + {Effect: External, Capacities: []Capacity{capacity("total")}}, + {Effect: CapacityOnly, Capacities: []Capacity{capacity("total"), capacity("total")}}, + {Effect: CapacityOnly, Capacities: []Capacity{{ID: "total", Limit: 1, Units: 2}}}, + {Effect: CapacityOnly, Capacities: []Capacity{{ID: "total", Units: 1}}}, + } { + if _, err := c.Acquire(t.Context(), request); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("request %+v: %v", request, err) + } + } +} diff --git a/internal/state/mutation/capacity_test.go b/internal/state/mutation/capacity_test.go new file mode 100644 index 0000000000..10e6f363da --- /dev/null +++ b/internal/state/mutation/capacity_test.go @@ -0,0 +1,330 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "reasonix/internal/state/workspacelease" +) + +func capacity(id string) Capacity { return Capacity{ID: id, Limit: 1, Units: 1} } + +func TestDisjointWriterCapacityWaitReportsActualBlockingResource(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + first := intent(filepath.Join(root, "first")) + first.Capacities = []Capacity{capacity("writers")} + held, err := c.Acquire(t.Context(), first) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + var notices []Notice + waiter, err := New(root, state, func(n Notice) { + if n.State != "released" { + notices = append(notices, n) + } + if n.State == "waiting" { + held.End(NotExecuted) + } + }) + if err != nil { + t.Fatal(err) + } + second := intent(filepath.Join(root, "second")) + second.Capacities = first.Capacities + lease, err := waiter.Acquire(t.Context(), second) + if err != nil { + t.Fatal(err) + } + lease.End(NotExecuted) + if len(notices) != 2 { + t.Fatalf("unpaired wait: %+v", notices) + } + for _, n := range notices { + if n.Effect != NamedPaths || n.BlockedOn != CapacityOnly { + t.Fatalf("request kind confused with blocking cause: %+v", n) + } + } +} + +func awaitCapacityNotice(t *testing.T, notices <-chan Notice, state string) Notice { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + for { + select { + case n := <-notices: + if n.State == state { + return n + } + case <-ctx.Done(): + t.Fatalf("missing %s notice", state) + } + } +} + +func TestWaitingFileTransactionDoesNotConsumeCapacity(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + owner := testCoordinator(t, root, state) + path := filepath.Join(root, "a") + held, err := owner.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + notices := make(chan Notice, 16) + waiter, err := New(root, state, func(n Notice) { notices <- n }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + request := intent(path) + request.Capacities = []Capacity{capacity("root-total")} + lease, err := waiter.Acquire(ctx, request) + if lease != nil { + lease.End(NotExecuted) + } + done <- err + }() + awaitCapacityNotice(t, notices, "waiting") + model, err := owner.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("root-total")}}) + if err != nil { + t.Fatalf("waiting writer consumed model capacity: %v", err) + } + defer model.End(NotExecuted) + held.End(NotExecuted) + model.End(NotExecuted) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("writer did not resume") + } +} + +func TestWaitingCapacityTransactionDoesNotHoldPaths(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + owner := testCoordinator(t, root, state) + held, err := owner.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("root-total")}}) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + notices := make(chan Notice, 16) + waiter, err := New(root, state, func(n Notice) { notices <- n }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + path := filepath.Join(root, "a") + go func() { + request := intent(path) + request.Capacities = []Capacity{capacity("root-total")} + lease, err := waiter.Acquire(ctx, request) + if lease != nil { + lease.End(NotExecuted) + } + done <- err + }() + awaitCapacityNotice(t, notices, "waiting") + // An unrelated writer has no admission to this root's execution budget. + other, err := testCoordinator(t, t.TempDir(), state).Acquire(t.Context(), intent(filepath.Join(root, "b"))) + if err != nil { + t.Fatal(err) + } + other.End(NotExecuted) + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + lease, err := owner.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatalf("cancelled capacity request retained paths: %v", err) + } + lease.End(NotExecuted) +} + +func TestCapacityGrantsReadyWaitersInOrder(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + owner := testCoordinator(t, root, state) + held, err := owner.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + granted := make(chan string, 2) + releases := []chan struct{}{make(chan struct{}), make(chan struct{})} + done := make(chan error, 2) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + for index, id := range []string{"first", "second"} { + notices := make(chan Notice, 16) + c, err := New(root, state, func(n Notice) { notices <- n }) + if err != nil { + t.Fatal(err) + } + go func() { + l, err := c.Acquire(ctx, Intent{Effect: CapacityOnly, ExecutionID: id, Capacities: []Capacity{capacity("total")}}) + if err == nil { + granted <- id + select { + case <-releases[index]: + case <-ctx.Done(): + } + l.End(NotExecuted) + } + done <- err + }() + awaitCapacityNotice(t, notices, "waiting") + } + held.End(NotExecuted) + select { + case id := <-granted: + if id != "first" { + t.Fatal(id) + } + case <-time.After(5 * time.Second): + t.Fatal("first waiter stuck") + } + close(releases[0]) + select { + case id := <-granted: + if id != "second" { + t.Fatal(id) + } + case <-time.After(5 * time.Second): + t.Fatal("second waiter stuck") + } + close(releases[1]) + for range 2 { + if err := <-done; err != nil { + t.Fatal(err) + } + } +} + +func TestCapacityIdentityCannotChangeLimit(t *testing.T) { + c := testCoordinator(t, t.TempDir(), t.TempDir()) + l, err := c.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if err != nil { + t.Fatal(err) + } + defer l.End(NotExecuted) + _, err = c.Acquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{{ID: "total", Limit: 2, Units: 1}}}) + if !errors.Is(err, ErrProtocol) { + t.Fatalf("capacity limit changed under an active owner: %v", err) + } +} + +func TestLegacyWaitDoesNotConsumeExecutionCapacity(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "a") + legacy, err := workspacelease.New(root, state, nil) + if err != nil { + t.Fatal(err) + } + legacy.BeginRun() + defer legacy.EndRun() + if err := legacy.AcquirePaths(t.Context(), []string{path}); err != nil { + t.Fatal(err) + } + notices := make(chan Notice, 16) + c, err := New(root, state, func(n Notice) { notices <- n }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + defer func() { cancel(); <-done }() + go func() { + req := intent(path) + req.Capacities = []Capacity{capacity("total")} + l, err := c.Acquire(ctx, req) + if l != nil { + l.End(NotExecuted) + } + done <- err + }() + awaitCapacityNotice(t, notices, "waiting") + probeCtx, stop := context.WithTimeout(t.Context(), time.Second) + defer stop() + l, err := testCoordinator(t, root, state).Acquire(probeCtx, Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("total")}}) + if err != nil { + t.Fatalf("legacy file wait exhausted execution capacity: %v", err) + } + l.End(NotExecuted) +} + +func TestLargeCapacityRequestCannotBeStarvedBySmallerLaterRequests(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + one := Intent{Effect: CapacityOnly, Capacities: []Capacity{{ID: "total", Limit: 2, Units: 1}}} + owner := testCoordinator(t, root, state) + held, err := owner.Acquire(t.Context(), one) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + type result struct { + lease *Lease + err error + } + start := func(request Intent) (<-chan Notice, <-chan result) { + notices, done := make(chan Notice, 16), make(chan result, 1) + c, err := New(root, state, func(n Notice) { notices <- n }) + if err != nil { + t.Fatal(err) + } + go func() { l, err := c.Acquire(ctx, request); done <- result{l, err} }() + return notices, done + } + largeNotices, largeDone := start(Intent{Effect: CapacityOnly, Capacities: []Capacity{{ID: "total", Limit: 2, Units: 2}}}) + awaitCapacityNotice(t, largeNotices, "waiting") + smallNotices, smallDone := start(one) + select { + case n := <-smallNotices: + if n.State != "waiting" { + t.Fatalf("unexpected event: %+v", n) + } + case got := <-smallDone: + if got.lease != nil { + got.lease.End(NotExecuted) + } + t.Fatalf("small request bypassed earlier large request: %v", got.err) + case <-ctx.Done(): + t.Fatal("small request did not queue") + } + held.End(NotExecuted) + var large result + select { + case large = <-largeDone: + case <-ctx.Done(): + t.Fatal("large request starved") + } + if large.err != nil { + t.Fatal(large.err) + } + defer large.lease.End(NotExecuted) + large.lease.End(NotExecuted) + select { + case small := <-smallDone: + if small.err != nil { + t.Fatal(small.err) + } + small.lease.End(NotExecuted) + case <-ctx.Done(): + t.Fatal("small request did not resume") + } +} diff --git a/internal/state/mutation/case_fold_test.go b/internal/state/mutation/case_fold_test.go new file mode 100644 index 0000000000..91df1ef758 --- /dev/null +++ b/internal/state/mutation/case_fold_test.go @@ -0,0 +1,39 @@ +package mutation + +import ( + "os" + "path/filepath" + "testing" +) + +func TestHeldPathsCoverOnlyTheSameFileAcrossCaseSpellings(t *testing.T) { + root := t.TempDir() + c, err := New(root, t.TempDir(), nil) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "existing.go"), nil, 0600); err != nil { + t.Fatal(err) + } + lease, err := c.Acquire(t.Context(), Intent{Effect: NamedPaths, Paths: []string{"absent.go", "existing.go"}}) + if err != nil { + t.Fatal(err) + } + defer lease.End(NotExecuted) + if err := lease.CheckPaths([]string{"ABSENT.go"}); err == nil { + t.Fatal("a different spelling of an absent file was treated as held") + } + other := filepath.Join(root, "EXISTING.go") + if info, statErr := os.Stat(other); statErr == nil { + held, _ := os.Stat(filepath.Join(root, "existing.go")) + if os.SameFile(info, held) { + if err := lease.CheckPaths([]string{"EXISTING.go"}); err != nil { + t.Fatalf("the same file under another spelling was refused: %v", err) + } + return + } + } + if err := lease.CheckPaths([]string{"EXISTING.go"}); err == nil { + t.Fatal("a distinct file was treated as held") + } +} diff --git a/internal/state/mutation/compatibility_phase_test.go b/internal/state/mutation/compatibility_phase_test.go new file mode 100644 index 0000000000..72f1c9caa9 --- /dev/null +++ b/internal/state/mutation/compatibility_phase_test.go @@ -0,0 +1,131 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "slices" + "testing" + + "reasonix/internal/base/filelock" + "reasonix/internal/state/workspacelease" +) + +func TestCompatibilityPreparationDoesNotHoldTheGlobalRegistry(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + first := testCoordinator(t, root, state) + a, b := filepath.Join(root, "a"), filepath.Join(root, "b") + entered, resume := make(chan struct{}), make(chan struct{}) + first.compatClaim = func(root, dir, session string, paths []string) (func(), *workspacelease.ConflictError, error) { + if slices.ContainsFunc(paths, func(path string) bool { return filepath.Base(path) == "a" }) { + close(entered) + <-resume + } + return workspacelease.TryClaimPaths(root, dir, session, paths) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + lease, err := first.Acquire(ctx, intent(a)) + if lease != nil { + lease.End(NotExecuted) + } + done <- err + }() + select { + case <-entered: + case err := <-done: + t.Fatalf("call returned before compatibility preparation: %v", err) + } + defer func() { close(resume); <-done }() + unlock, err := filelock.TryAcquire(first.guardPath()) + if err != nil { + t.Fatalf("compatibility discovery retained the global registry: %v", err) + } + unlock() + second := testCoordinator(t, root, state) + lease, err := second.Acquire(t.Context(), intent(b)) + if err != nil { + t.Fatalf("unrelated resources blocked by compatibility preparation: %v", err) + } + lease.End(NotExecuted) +} + +func TestPreparingCompatibilityClaimRetainsCapacityQueuePriority(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + first := testCoordinator(t, root, state) + claimed, resume := make(chan struct{}), make(chan struct{}) + first.compatClaim = func(root, dir, session string, paths []string) (func(), *workspacelease.ConflictError, error) { + release, conflict, err := workspacelease.TryClaimPaths(root, dir, session, paths) + if err == nil { + close(claimed) + <-resume + } + return release, conflict, err + } + pool := []Capacity{{ID: "writers", Limit: 1, Units: 1}} + want := intent(filepath.Join(root, "a")) + want.Capacities = pool + done := make(chan error, 1) + go func() { + lease, err := first.Acquire(t.Context(), want) + if lease != nil { + lease.End(NotExecuted) + } + done <- err + }() + select { + case <-claimed: + case err := <-done: + t.Fatalf("first call returned before compatibility grant: %v", err) + } + defer func() { + close(resume) + if err := <-done; err != nil { + t.Errorf("first call lost admission: %v", err) + } + }() + second := testCoordinator(t, root, state) + want = intent(filepath.Join(root, "b")) + want.Capacities = pool + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + second.notify = func(notice Notice) { + if notice.State == "waiting" { + cancel() + } + } + lease, err := second.Acquire(ctx, want) + if lease != nil { + lease.End(NotExecuted) + t.Fatal("later disjoint path stole the preparing transaction's capacity") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("later capacity waiter: %v", err) + } +} + +func TestCancellationAfterCompatibilityGrantLeavesNoClaims(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + path := filepath.Join(root, "a") + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + c.compatClaim = func(root, dir, session string, paths []string) (func(), *workspacelease.ConflictError, error) { + release, conflict, err := workspacelease.TryClaimPaths(root, dir, session, paths) + if err == nil { + cancel() + } + return release, conflict, err + } + if _, err := c.Acquire(ctx, intent(path)); !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled compatibility admission: %v", err) + } + second := testCoordinator(t, root, state) + lease, err := second.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatalf("cancelled admission retained new claims: %v", err) + } + lease.End(NotExecuted) +} diff --git a/internal/state/mutation/compatibility_test.go b/internal/state/mutation/compatibility_test.go new file mode 100644 index 0000000000..20e888a531 --- /dev/null +++ b/internal/state/mutation/compatibility_test.go @@ -0,0 +1,27 @@ +package mutation + +import ( + "errors" + "os" + "path/filepath" + "testing" +) + +func TestUnknownOrCorruptCoordinationRecordIsPreserved(t *testing.T) { + for _, raw := range []string{`{"version":2,"id":"future","order":1,"paths":["/future"]}`, `{"version":`} { + t.Run(raw, func(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + path := c.recordPath("future") + if err := os.WriteFile(path, []byte(raw), 0600); err != nil { + t.Fatal(err) + } + if _, err := c.Acquire(t.Context(), intent(filepath.Join(root, "file"))); !errors.Is(err, ErrProtocol) { + t.Fatalf("acquire: %v", err) + } + if got, err := os.ReadFile(path); err != nil || string(got) != raw { + t.Fatalf("unknown record modified: %q %v", got, err) + } + }) + } +} diff --git a/internal/state/mutation/completion.go b/internal/state/mutation/completion.go new file mode 100644 index 0000000000..5eca07ffb4 --- /dev/null +++ b/internal/state/mutation/completion.go @@ -0,0 +1,86 @@ +package mutation + +import ( + "context" + "log/slog" + "os" + "sync" + "time" + + "reasonix/internal/base/filelock" +) + +type leaseCompletion struct { + mu sync.Mutex + waits []<-chan struct{} +} + +func HoldUntil(ctx context.Context, done <-chan struct{}) error { + l, _ := ctx.Value(leaseContextKey{}).(*Lease) + return l.HoldUntil(done) +} + +// HoldUntil retains physical ownership after End until execution is settled. +func (l *Lease) HoldUntil(done <-chan struct{}) error { + if l == nil || l.coordinator == nil || done == nil { + return nil + } + l.completion.mu.Lock() + defer l.completion.mu.Unlock() + if l.closed.Load() { + return ErrInvalidIntent + } + l.completion.waits = append(l.completion.waits, done) + return nil +} + +func (l *Lease) endAfterCompletion(outcome Outcome) { + l.completion.mu.Lock() + waits := l.completion.waits + l.completion.waits = nil + l.completion.mu.Unlock() + for i, done := range waits { + select { + case <-done: + default: + go func() { + for _, done := range waits[i:] { + <-done + } + l.releaseResources(outcome) + }() + return + } + } + l.releaseResources(outcome) +} + +func (l *Lease) releaseResources(outcome Outcome) { + if l.coordinator == nil { + return + } + c := l.coordinator + if l.releaseCompat != nil { + l.releaseCompat() + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if unlock, err := filelock.Acquire(ctx, c.guardPath()); err == nil { + _ = os.Remove(c.recordPath(l.ID)) + if l.releaseLive != nil { + l.releaseLive() + l.releaseLive = nil + } + _ = os.Remove(c.livePath(l.ID)) + _ = os.Remove(c.preparationPath(l.ID)) + unlock() + } + if l.releaseLive != nil { + l.releaseLive() + } + if !l.acquiredAt.IsZero() { + if err := c.notice(l.record, "released", nil, outcome); err != nil { + slog.Warn("mutation release notification failed after resource release", "transaction", l.ID, "err", err) + } + } +} diff --git a/internal/state/mutation/completion_test.go b/internal/state/mutation/completion_test.go new file mode 100644 index 0000000000..a583cb760c --- /dev/null +++ b/internal/state/mutation/completion_test.go @@ -0,0 +1,55 @@ +package mutation + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestEndedTransactionRetainsResourcesUntilAllExecutionsSettle(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + first, second := testCoordinator(t, root, state), testCoordinator(t, root, state) + l, err := first.Acquire(t.Context(), Intent{Effect: Workspace}) + if err != nil { + t.Fatal(err) + } + a, b := make(chan struct{}), make(chan struct{}) + if err := HoldUntil(l.Context(t.Context()), a); err != nil { + t.Fatal(err) + } + if err := l.HoldUntil(b); err != nil { + t.Fatal(err) + } + l.End(Unknown) + if l.HoldingPaths() || l.CheckPaths([]string{root}) == nil { + t.Fatal("ended caller retained execution authority") + } + if err := l.HoldUntil(make(chan struct{})); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("ended caller registered new execution: %v", err) + } + assertBusy := func() { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond) + defer cancel() + other, err := second.Acquire(ctx, Intent{Effect: Workspace}) + if other != nil { + other.End(NotExecuted) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("unsettled execution lost exclusion: %v", err) + } + } + assertBusy() + close(a) + assertBusy() + close(b) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + next, err := second.Acquire(ctx, Intent{Effect: Workspace}) + if err != nil { + t.Fatal(err) + } + defer next.End(NotExecuted) + l.End(NotExecuted) + assertBusy() +} diff --git a/internal/state/mutation/continuation.go b/internal/state/mutation/continuation.go new file mode 100644 index 0000000000..470a09cf5c --- /dev/null +++ b/internal/state/mutation/continuation.go @@ -0,0 +1,37 @@ +package mutation + +import ( + "context" + "slices" +) + +// Continuation admits asynchronous work as a new transaction after its caller +// releases the current one. It carries resource identity, never a live lease. +type Continuation struct { + coordinator *Coordinator + intent Intent +} + +func (l *Lease) PrepareContinuation() (*Continuation, error) { + if l == nil || l.coordinator == nil { + return nil, nil + } + if l.closed.Load() { + return nil, ErrInvalidIntent + } + effect := l.record.Effect + if effect == "" { + effect = NamedPaths + } + return &Continuation{coordinator: l.coordinator, intent: Intent{Effect: effect, + Capacities: slices.Clone(l.record.Capacities), + Paths: slices.Clone(l.record.Paths), SessionID: l.record.SessionID, + ExecutionID: l.record.ExecutionID, CallID: l.record.CallID}}, nil +} + +func (c *Continuation) Acquire(ctx context.Context) (*Lease, error) { + if c == nil { + return nil, nil + } + return c.coordinator.Acquire(WithoutLease(ctx), c.intent) +} diff --git a/internal/state/mutation/coordinator.go b/internal/state/mutation/coordinator.go new file mode 100644 index 0000000000..75d5637554 --- /dev/null +++ b/internal/state/mutation/coordinator.go @@ -0,0 +1,280 @@ +package mutation + +import ( + "context" + "crypto/rand" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "sync" + "sync/atomic" + "time" + + "reasonix/internal/base/filelock" + "reasonix/internal/base/fileutil" + "reasonix/internal/state/workspacelease" +) + +const DefaultWaitLimit = 30 * time.Second +const retryInterval = 20 * time.Millisecond + +type Coordinator struct { + root string + dir string + compatDir string + notify func(Notice) + admit func(context.Context, Intent) error + workspaceResources []string + compatClaim func(string, string, string, []string) (func(), *workspacelease.ConflictError, error) +} + +type Options struct { + AdmissionCheck func(context.Context, Intent) error + WorkspaceResources []string +} + +func New(root, stateDir string, notify func(Notice), opts ...Options) (*Coordinator, error) { + canonical, err := fileutil.CanonicalWritePath(root) + if err != nil { + return nil, err + } + if stateDir == "" { + return nil, ErrInvalidIntent + } + dir := filepath.Join(stateDir, "mutations-v1") + if err := os.MkdirAll(dir, 0700); err != nil { + return nil, err + } + c := &Coordinator{root: canonical, dir: dir, compatDir: stateDir, notify: notify, compatClaim: workspacelease.TryClaimPaths} + for _, opt := range opts { + c.admit = opt.AdmissionCheck + c.workspaceResources = slices.Clone(opt.WorkspaceResources) + } + return c, nil +} + +type Lease struct { + ID string + once sync.Once + coordinator *Coordinator + record record + releaseLive func() + releaseCompat func() + acquiredAt time.Time + closed atomic.Bool + completion leaseCompletion +} + +// TryAcquire preserves queue order but never waits for a conflicting resource. +// Registry operations remain cancellable short critical sections. +func (c *Coordinator) TryAcquire(ctx context.Context, intent Intent) (*Lease, error) { + intent.tryOnly = true + return c.Acquire(ctx, intent) +} + +func (c *Coordinator) Acquire(ctx context.Context, intent Intent) (acquired *Lease, resultErr error) { + var owned *Lease + waiting := false + defer func() { + if p := recover(); p != nil { + acquired, resultErr = nil, fmt.Errorf("%w: admission panicked: %v", ErrCallback, p) + } + resultErr = c.abandonAcquisition(owned, acquired, waiting, resultErr) + }() + if err := ctx.Err(); err != nil { + return nil, err + } + capacities, err := normalizedCapacities(intent.Capacities) + if err != nil { + return nil, err + } + intent.Capacities = capacities + paths, err := c.resources(intent) + if err != nil { + return nil, err + } + if len(paths) == 0 && len(capacities) == 0 { + return &Lease{}, nil + } + if parent, _ := ctx.Value(leaseContextKey{}).(*Lease); parent != nil && parent.coordinator == c { + if parent.closed.Load() || !covers(parent.record.Paths, paths) || !coversCapacities(parent.record.Capacities, capacities) { + return nil, ErrInvalidIntent + } + if err := c.validate(ctx, intent, paths); err != nil { + return nil, err + } + return &Lease{}, nil + } + ctx, cancel := acquisitionContext(ctx, intent.Effect) + defer cancel() + l := &Lease{ID: rand.Text(), coordinator: c, record: record{ + Version: 1, SessionID: intent.SessionID, ExecutionID: intent.ExecutionID, + CallID: intent.CallID, Paths: paths, StartedAt: time.Now(), + Effect: intent.Effect, Capacities: capacities, + Workspace: c.root, + }} + if len(capacities) != 0 { + l.record.Version = 2 + } + l.record.ID = l.ID + defer func() { + if errors.Is(resultErr, context.DeadlineExceeded) { + resultErr = &WaitError{TransactionID: l.ID, Paths: paths, Cause: resultErr, Effect: intent.Effect, Capacities: capacities} + } + }() + l.releaseLive, err = filelock.TryAcquire(c.livePath(l.ID)) + if err != nil { + return nil, err + } + owned = l + if err := c.enqueue(ctx, &l.record); err != nil { + return nil, err + } + probe := versionProbe{} + for { + if err := ctx.Err(); err != nil { + return nil, err + } + if err := probe.check(ctx, intent.Expected, paths); err != nil { + return nil, err + } + held, err := c.tryGrant(ctx, l) + if err != nil { + return nil, err + } + if held == nil { + return c.finishAcquisition(ctx, l, intent, paths, waiting) + } + if intent.tryOnly { + return nil, ErrBusy + } + if !waiting { + waiting = true + l.record.waitKind = held.kind + if err := c.notice(l.record, "waiting", &held.record, NotExecuted); err != nil { + return nil, err + } + } + timer := time.NewTimer(retryInterval) + select { + case <-ctx.Done(): + timer.Stop() + case <-timer.C: + } + } +} + +func (c *Coordinator) finishAcquisition(ctx context.Context, lease *Lease, intent Intent, paths []string, waiting bool) (*Lease, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if err := c.validate(ctx, intent, paths); err != nil { + return nil, err + } + lease.acquiredAt = time.Now() + if waiting { + if err := c.notice(lease.record, "acquired", nil, NotExecuted); err != nil { + return nil, err + } + } + return lease, nil +} + +func (c *Coordinator) abandonAcquisition(owned, acquired *Lease, waiting bool, err error) error { + if owned == nil || acquired != nil { + return err + } + owned.End(NotExecuted) + if waiting { + return errors.Join(err, c.waitEnded(owned.record, err)) + } + return err +} + +func (l *Lease) End(outcome Outcome) { + if l == nil { + return + } + l.once.Do(func() { + l.closed.Store(true) + l.endAfterCompletion(outcome) + }) +} + +type leaseContextKey struct{} + +func (l *Lease) HoldingPaths() bool { + return l != nil && l.coordinator != nil && len(l.record.Paths) != 0 && !l.closed.Load() +} + +func (l *Lease) CheckPaths(paths []string) error { + if l == nil || l.coordinator == nil || l.closed.Load() { + return ErrInvalidIntent + } + requested, err := l.coordinator.resources(Intent{Effect: NamedPaths, Paths: paths}) + if err != nil { + return err + } + if !covers(l.record.Paths, requested) { + return ErrInvalidIntent + } + return nil +} + +func (l *Lease) Context(ctx context.Context) context.Context { + if l == nil || l.coordinator == nil { + return ctx + } + return context.WithValue(ctx, leaseContextKey{}, l) +} + +func WithoutLease(ctx context.Context) context.Context { + return context.WithValue(ctx, leaseContextKey{}, (*Lease)(nil)) +} + +func (c *Coordinator) notice(r record, state string, holder *record, outcome Outcome) error { + return c.emitNotice(r, state, "", holder, outcome) +} + +func (c *Coordinator) emitNotice(r record, state, reason string, holder *record, outcome Outcome) (resultErr error) { + defer func() { + if p := recover(); p != nil { + resultErr = fmt.Errorf("%w: notice %s panicked: %v", ErrCallback, state, p) + } + }() + if c.notify == nil { + return nil + } + n := Notice{TransactionID: r.ID, SessionID: r.SessionID, ExecutionID: r.ExecutionID, + Effect: r.Effect, BlockedOn: r.waitKind, Capacities: slices.Clone(r.Capacities), + CallID: r.CallID, Paths: slices.Clone(r.Paths), State: state, StartedAt: r.StartedAt, + Elapsed: time.Since(r.StartedAt), Outcome: outcome, Reason: reason} + if holder != nil { + n.HolderSessionID, n.HolderExecutionID = holder.SessionID, holder.ExecutionID + n.HolderTransactionID, n.HolderPaths = holder.ID, slices.Clone(holder.Paths) + } + c.notify(n) + return nil +} + +func (c *Coordinator) waitEnded(r record, err error) error { + reason := "rejected" + switch { + case errors.Is(err, context.Canceled): + reason = "cancelled" + case errors.Is(err, context.DeadlineExceeded): + reason = "timeout" + case errors.Is(err, ErrVersionConflict): + reason = "conflict" + } + return c.emitNotice(r, "abandoned", reason, nil, NotExecuted) +} + +func (c *Coordinator) recordPath(id string) string { return filepath.Join(c.dir, id+".json") } +func (c *Coordinator) livePath(id string) string { return filepath.Join(c.dir, id+".live") } +func (c *Coordinator) guardPath() string { return filepath.Join(c.dir, "registry.lock") } +func (c *Coordinator) preparationPath(id string) string { + return filepath.Join(c.dir, id+".preparing") +} diff --git a/internal/state/mutation/coordinator_test.go b/internal/state/mutation/coordinator_test.go new file mode 100644 index 0000000000..a1e8012756 --- /dev/null +++ b/internal/state/mutation/coordinator_test.go @@ -0,0 +1,172 @@ +package mutation + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" +) + +func testCoordinator(t *testing.T, root, state string) *Coordinator { + t.Helper() + c, err := New(root, state, nil) + if err != nil { + t.Fatal(err) + } + return c +} + +func intent(paths ...string) Intent { + return Intent{Effect: NamedPaths, Paths: paths, SessionID: "test", ExecutionID: "execution", CallID: "call"} +} + +func TestTransactionReleasesBeforeSessionEnds(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + a, b := testCoordinator(t, root, state), testCoordinator(t, root, state) + first, err := a.Acquire(t.Context(), intent(filepath.Join(root, "a"))) + if err != nil { + t.Fatal(err) + } + second, err := b.Acquire(t.Context(), intent(filepath.Join(root, "b"))) + if err != nil { + t.Fatal(err) + } + second.End(NotExecuted) + ctx, cancel := context.WithTimeout(t.Context(), 40*time.Millisecond) + defer cancel() + if _, err := b.Acquire(ctx, intent(filepath.Join(root, "a"))); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("overlap: %v", err) + } + first.End(Committed) + next, err := b.Acquire(t.Context(), intent(filepath.Join(root, "a"))) + if err != nil { + t.Fatal(err) + } + defer next.End(NotExecuted) + first.End(NotExecuted) + ctx2, cancel2 := context.WithTimeout(t.Context(), 40*time.Millisecond) + defer cancel2() + if _, err := a.Acquire(ctx2, intent(filepath.Join(root, "a"))); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("old release removed new lease: %v", err) + } +} + +func TestDifferentWorkspacesCoordinateExtraPathAndHardlinks(t *testing.T) { + state, extra := t.TempDir(), filepath.Join(t.TempDir(), "shared") + if err := os.WriteFile(extra, []byte("base"), 0600); err != nil { + t.Fatal(err) + } + a, b := testCoordinator(t, t.TempDir(), state), testCoordinator(t, t.TempDir(), state) + lease, err := a.Acquire(t.Context(), intent(extra)) + if err != nil { + t.Fatal(err) + } + defer lease.End(NotExecuted) + alias := filepath.Join(t.TempDir(), "alias") + if err := os.Link(extra, alias); err != nil { + t.Skipf("hardlinks unavailable: %v", err) + } + for _, path := range []string{extra, alias} { + ctx, cancel := context.WithTimeout(t.Context(), 40*time.Millisecond) + _, err := b.Acquire(ctx, intent(path)) + cancel() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("%s: %v", path, err) + } + } +} + +func TestQueuedSetDoesNotHoldPartialResources(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + a, b := filepath.Join(root, "a"), filepath.Join(root, "b") + first, err := c.Acquire(t.Context(), intent(a)) + if err != nil { + t.Fatal(err) + } + defer first.End(NotExecuted) + queued := make(chan Notice, 8) + waiter, err := New(root, state, func(n Notice) { queued <- n }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + l, e := waiter.Acquire(ctx, intent(a, b)) + if l != nil { + l.End(NotExecuted) + } + done <- e + }() + select { + case <-queued: + case <-time.After(3 * time.Second): + t.Fatal("wait was not reported") + } + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + last, err := c.Acquire(t.Context(), intent(b)) + if err != nil { + t.Fatal(err) + } + last.End(NotExecuted) +} + +func TestExplicitEffectsAndContentPreconditions(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + if _, err := c.Acquire(t.Context(), intent()); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("empty named set: %v", err) + } + path := filepath.Join(root, "a") + if err := os.WriteFile(path, []byte("before"), 0600); err != nil { + t.Fatal(err) + } + version, err := ReadVersionContext(t.Context(), path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("after"), 0600); err != nil { + t.Fatal(err) + } + i := intent(path) + i.Expected = map[string]ContentVersion{path: version} + if _, err := c.Acquire(t.Context(), i); !errors.Is(err, ErrVersionConflict) { + t.Fatalf("stale precondition: %v", err) + } + lease, err := c.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatalf("failed precondition leaked resources: %v", err) + } + lease.End(NotExecuted) +} + +func TestNestedTransactionCannotWidenOrReleaseParent(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + c := testCoordinator(t, root, state) + path := filepath.Join(root, "a") + parent, err := c.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatal(err) + } + defer parent.End(NotExecuted) + ctx := parent.Context(t.Context()) + child, err := c.Acquire(ctx, intent(path)) + if err != nil { + t.Fatal(err) + } + child.End(NotExecuted) + if _, err := c.Acquire(ctx, intent(filepath.Join(root, "b"))); !errors.Is(err, ErrInvalidIntent) { + t.Fatalf("nested scope widened: %v", err) + } + other, cancel := context.WithTimeout(t.Context(), 40*time.Millisecond) + defer cancel() + if _, err := c.Acquire(other, intent(path)); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("child released parent: %v", err) + } +} diff --git a/internal/state/mutation/doc.go b/internal/state/mutation/doc.go new file mode 100644 index 0000000000..894ae6a47b --- /dev/null +++ b/internal/state/mutation/doc.go @@ -0,0 +1,11 @@ +// Package mutation admits complete resource sets for short write transactions. +// It coordinates cooperating processes; it does not grant filesystem access, +// execute tools, or provide a sandbox. Callers authorize the complete effect +// before acquisition, recheck their own preconditions, perform the mutation, +// and end the lease after physical execution settles. +// +// Coordinators that can touch the same files must share a trusted state +// directory. The registry serializes admission, not execution. Live OS locks +// distinguish a queued or active transaction from a dead owner. Compatibility +// claims retain exclusion against older workspacelease writers. +package mutation diff --git a/internal/state/mutation/fairness_test.go b/internal/state/mutation/fairness_test.go new file mode 100644 index 0000000000..105940b80a --- /dev/null +++ b/internal/state/mutation/fairness_test.go @@ -0,0 +1,67 @@ +package mutation + +import ( + "context" + "path/filepath" + "testing" + "time" +) + +func TestOverlappingQueueIsFIFOAndDisjointWorkProceeds(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + coordinator := testCoordinator(t, root, state) + a, b, c := filepath.Join(root, "a"), filepath.Join(root, "b"), filepath.Join(root, "c") + held, err := coordinator.Acquire(t.Context(), intent(a)) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + type result struct { + name string + lease *Lease + err error + } + done := make(chan result, 2) + start := func(name string, paths ...string) { + queued := make(chan struct{}, 1) + waiter, err := New(root, state, func(n Notice) { + if n.State == "waiting" { + queued <- struct{}{} + } + }) + if err != nil { + t.Fatal(err) + } + go func() { lease, err := waiter.Acquire(ctx, intent(paths...)); done <- result{name, lease, err} }() + select { + case <-queued: + case <-ctx.Done(): + t.Fatal("waiter did not enter queue") + } + } + start("first", a, b) + start("second", a) + disjoint, err := coordinator.Acquire(ctx, intent(c)) + if err != nil { + t.Fatalf("disjoint work blocked: %v", err) + } + disjoint.End(NotExecuted) + held.End(NotExecuted) + first := <-done + if first.lease != nil { + defer first.lease.End(NotExecuted) + } + if first.err != nil || first.name != "first" { + t.Fatalf("queue head: %+v", first) + } + first.lease.End(NotExecuted) + second := <-done + if second.lease != nil { + defer second.lease.End(NotExecuted) + } + if second.err != nil || second.name != "second" { + t.Fatalf("queue successor: %+v", second) + } +} diff --git a/internal/state/mutation/notice_test.go b/internal/state/mutation/notice_test.go new file mode 100644 index 0000000000..ffd67f28bb --- /dev/null +++ b/internal/state/mutation/notice_test.go @@ -0,0 +1,117 @@ +package mutation + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "testing" + + "reasonix/internal/state/workspacelease" +) + +func TestLegacyWaitClosesWithoutPrematureAcquisition(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + legacy, err := workspacelease.New(root, state, nil) + if err != nil { + t.Fatal(err) + } + legacy.BeginRun() + defer legacy.EndRun() + path := filepath.Join(root, "a") + if err := legacy.AcquirePaths(t.Context(), []string{path}); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var states []string + c, err := New(root, state, func(n Notice) { + states = append(states, n.State) + if n.State == "waiting" { + cancel() + } + }) + if err != nil { + t.Fatal(err) + } + if _, err := c.Acquire(ctx, intent(path)); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if !slices.Equal(states, []string{"waiting", "abandoned"}) { + t.Fatalf("events: %v", states) + } + legacy.EndRun() + lease, err := c.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatalf("refusal retained a resource: %v", err) + } + lease.End(NotExecuted) +} + +func TestWaitNoticesRetainIdentityAndTerminationReason(t *testing.T) { + for _, finish := range []string{"acquired", "cancelled", "conflict"} { + t.Run(finish, func(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "a") + holder := testCoordinator(t, root, state) + held, err := holder.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var notices []Notice + c, err := New(root, state, func(n Notice) { + if n.State == "released" { + return + } + notices = append(notices, n) + if n.State != "waiting" { + return + } + switch finish { + case "cancelled": + cancel() + case "conflict": + if err := os.WriteFile(path, []byte("changed"), 0600); err != nil { + t.Fatal(err) + } + held.End(Committed) + default: + held.End(NotExecuted) + } + }) + if err != nil { + t.Fatal(err) + } + request := intent(path) + request.Expected = map[string]ContentVersion{path: {Kind: "absent"}} + lease, err := c.Acquire(ctx, request) + if lease != nil { + lease.End(NotExecuted) + } + if (finish == "acquired") != (err == nil) { + t.Fatalf("acquire: %v", err) + } + if len(notices) != 2 { + t.Fatalf("notices: %+v", notices) + } + start, end := notices[0], notices[1] + if start.TransactionID == "" || end.TransactionID != start.TransactionID || end.StartedAt != start.StartedAt || start.State != "waiting" { + t.Fatalf("unpaired wait: %+v", notices) + } + if start.HolderTransactionID != held.ID || start.HolderExecutionID != "execution" || len(start.HolderPaths) != 1 { + t.Fatalf("missing holder: %+v", start) + } + if finish == "acquired" { + if end.State != "acquired" || end.Reason != "" { + t.Fatalf("end: %+v", end) + } + } else if end.State != "abandoned" || end.Reason != finish { + t.Fatalf("end: %+v", end) + } + }) + } +} diff --git a/internal/state/mutation/process_test.go b/internal/state/mutation/process_test.go new file mode 100644 index 0000000000..71844aa00d --- /dev/null +++ b/internal/state/mutation/process_test.go @@ -0,0 +1,77 @@ +package mutation + +import ( + "bufio" + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +func TestMutationProcessHelper(t *testing.T) { + if os.Getenv("REASONIX_MUTATION_TEST_HELPER") != "1" { + return + } + c, err := New(os.Getenv("REASONIX_MUTATION_TEST_ROOT"), os.Getenv("REASONIX_MUTATION_TEST_STATE"), nil) + if err != nil { + t.Fatal(err) + } + l, err := c.Acquire(t.Context(), intent(os.Getenv("REASONIX_MUTATION_TEST_PATH"))) + if err != nil { + t.Fatal(err) + } + defer l.End(NotExecuted) + fmt.Println("mutation-ready") + _, _ = bufio.NewReader(os.Stdin).ReadString('\n') +} + +func TestCrossProcessClaimAndCrashRecovery(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "a") + cmd := exec.Command(os.Args[0], "-test.run=^TestMutationProcessHelper$") + cmd.Env = append(os.Environ(), "REASONIX_MUTATION_TEST_HELPER=1", "REASONIX_MUTATION_TEST_ROOT="+root, + "REASONIX_MUTATION_TEST_STATE="+state, "REASONIX_MUTATION_TEST_PATH="+path) + in, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + out, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = in.Close(); _ = cmd.Process.Kill(); _ = cmd.Wait() }) + ready := make(chan string, 1) + go func() { line, _ := bufio.NewReader(out).ReadString('\n'); ready <- line }() + select { + case line := <-ready: + if line != "mutation-ready\n" { + t.Fatal(line) + } + case <-time.After(5 * time.Second): + t.Fatal("child did not acquire") + } + c := testCoordinator(t, root, state) + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + if _, err := c.Acquire(ctx, intent(path)); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cross-process overlap: %v", err) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = cmd.Wait() + ctx2, cancel2 := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel2() + l, err := c.Acquire(ctx2, intent(path)) + if err != nil { + t.Fatalf("crashed holder was not reclaimed: %v", err) + } + l.End(NotExecuted) +} diff --git a/internal/state/mutation/registry.go b/internal/state/mutation/registry.go new file mode 100644 index 0000000000..a140b8a98f --- /dev/null +++ b/internal/state/mutation/registry.go @@ -0,0 +1,202 @@ +package mutation + +import ( + "context" + "encoding/json" + "errors" + "math" + "os" + "path/filepath" + "strings" + "time" + + "reasonix/internal/base/filelock" + "reasonix/internal/base/fileutil" + "reasonix/internal/state/workspacelease" +) + +type record struct { + Version int `json:"version"` + ID string `json:"id"` + SessionID string `json:"sessionId"` + ExecutionID string `json:"executionId"` + CallID string `json:"callId"` + Paths []string `json:"paths"` + Order uint64 `json:"order"` + Active bool `json:"active"` + StartedAt time.Time `json:"startedAt"` + Effect Effect `json:"effect,omitempty"` + Capacities []Capacity `json:"capacities,omitempty"` + Workspace string `json:"workspace,omitempty"` + waitKind Effect +} + +type blocker struct { + record + kind Effect +} + +func (c *Coordinator) records() ([]record, error) { + entries, err := os.ReadDir(c.dir) + if err != nil { + return nil, err + } + var records []record + for _, entry := range entries { + if !strings.HasSuffix(entry.Name(), ".json") { + continue + } + id := strings.TrimSuffix(entry.Name(), ".json") + data, err := os.ReadFile(c.recordPath(id)) + if err != nil { + return nil, err + } + var r record + if err := json.Unmarshal(data, &r); err != nil { + return nil, errors.Join(ErrProtocol, err) + } + if r.ID != id || r.Order == 0 || !validRecordResources(r) { + return nil, ErrProtocol + } + for _, path := range r.Paths { + if !filepath.IsAbs(path) { + return nil, ErrProtocol + } + } + release, err := filelock.TryAcquire(c.livePath(id)) + if err == nil { + removeErr := os.Remove(c.recordPath(id)) + _ = os.Remove(c.preparationPath(id)) + release() + if removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { + return nil, removeErr + } + continue + } + if !errors.Is(err, filelock.ErrHeld) { + return nil, err + } + records = append(records, r) + } + return records, nil +} + +func (c *Coordinator) save(r record) error { + data, err := json.Marshal(r) + if err != nil { + return err + } + return fileutil.AtomicWriteFile(c.recordPath(r.ID), data, 0600) +} + +func (c *Coordinator) enqueue(ctx context.Context, r *record) error { + unlock, err := filelock.Acquire(ctx, c.guardPath()) + if err != nil { + return err + } + defer unlock() + records, err := c.records() + if err != nil { + return err + } + if _, err := capacityLimits(append(records, *r)); err != nil { + return err + } + r.Order = 1 + for _, other := range records { + if other.Order == math.MaxUint64 { + return ErrProtocol + } + r.Order = max(r.Order, other.Order+1) + } + return c.save(*r) +} + +func (c *Coordinator) tryGrant(ctx context.Context, lease *Lease) (*blocker, error) { + r := &lease.record + held, releasePreparation, err := c.grantBlocker(ctx, *r) + if held != nil || err != nil { + return held, err + } + defer releasePreparation() + var releaseCompat func() + defer func() { + if releaseCompat != nil { + releaseCompat() + } + }() + if len(r.Paths) != 0 { + release, conflict, err := c.compatClaim(c.root, c.compatDir, r.SessionID, r.Paths) + if errors.Is(err, workspacelease.ErrClaimBusy) { + held := &record{Paths: r.Paths} + if conflict != nil { + held.SessionID, held.Paths = conflict.SessionID, conflict.Paths + } + return &blocker{record: *held, kind: NamedPaths}, nil + } + if err != nil { + return nil, err + } + releaseCompat = release + if releaseCompat == nil { + return nil, ErrProtocol + } + } + unlock, err := filelock.Acquire(ctx, c.guardPath()) + if err != nil { + return nil, err + } + defer unlock() + records, err := c.records() + if err != nil { + return nil, err + } + if held, err := c.recordsBlocker(*r, records); held != nil || err != nil { + return held, err + } + r.Active = true + if err := c.save(*r); err != nil { + return nil, err + } + lease.releaseCompat, releaseCompat = releaseCompat, nil + return nil, nil +} + +// Queue order reserves eligibility while compatibility discovery runs outside +// the registry lock. The final locked recheck precedes active publication. +func (c *Coordinator) grantBlocker(ctx context.Context, r record) (*blocker, func(), error) { + unlock, err := filelock.Acquire(ctx, c.guardPath()) + if err != nil { + return nil, nil, err + } + defer unlock() + records, err := c.records() + if err != nil { + return nil, nil, err + } + if held, err := c.recordsBlocker(r, records); held != nil || err != nil { + return held, nil, err + } + release, err := filelock.TryAcquire(c.preparationPath(r.ID)) + return nil, release, err +} + +func (c *Coordinator) recordsBlocker(r record, records []record) (*blocker, error) { + if held := pathBlocker(r, records); held != nil { + return &blocker{record: *held, kind: NamedPaths}, nil + } + held, err := c.capacityBlocker(r, records) + if err != nil || held == nil { + return nil, err + } + return &blocker{record: *held, kind: CapacityOnly}, nil +} + +func pathBlocker(r record, records []record) *record { + for _, other := range records { + if other.ID != r.ID && (other.Active || other.Order < r.Order) && overlaps(r.Paths, other.Paths) { + return &other + } + } + return nil +} diff --git a/internal/state/mutation/resources.go b/internal/state/mutation/resources.go new file mode 100644 index 0000000000..817d3b00fb --- /dev/null +++ b/internal/state/mutation/resources.go @@ -0,0 +1,190 @@ +package mutation + +import ( + "context" + "fmt" + "os" + "path/filepath" + "runtime" + "slices" + "strings" + + "reasonix/internal/base/fileutil" +) + +func (c *Coordinator) resources(i Intent) ([]string, error) { + paths := i.Paths + switch i.Effect { + case CapacityOnly: + if len(paths) != 0 || len(i.Expected) != 0 || len(i.Capacities) == 0 || i.Recovery { + return nil, ErrInvalidIntent + } + return nil, nil + case ReadOnly, External: + if len(paths) != 0 || len(i.Expected) != 0 || len(i.Capacities) != 0 { + return nil, ErrInvalidIntent + } + return nil, nil + case Workspace: + paths = append([]string{c.root}, paths...) + paths = append(paths, c.workspaceResources...) + case NamedPaths, GitMetadata: + if len(paths) == 0 { + return nil, ErrInvalidIntent + } + default: + return nil, ErrInvalidIntent + } + out := make([]string, 0, len(paths)) + for _, p := range paths { + if strings.TrimSpace(p) == "" { + return nil, ErrInvalidIntent + } + if !filepath.IsAbs(p) { + p = filepath.Join(c.root, p) + } + canonical, err := fileutil.CanonicalWritePath(p) + if err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalidIntent, err) + } + out = append(out, canonical) + } + slices.Sort(out) + return slices.Compact(out), nil +} + +func within(root, path string) bool { + if runtime.GOOS == "windows" || runtime.GOOS == "darwin" { + root, path = strings.ToLower(root), strings.ToLower(path) + } + return withinExact(root, path) +} + +func withinExact(root, path string) bool { + rel, err := filepath.Rel(root, path) + return err == nil && (rel == "." || (rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)))) +} + +// heldWithin authorizes path under root. A case-folded spelling counts only +// when the filesystem resolves that prefix to root itself. +func heldWithin(root, path string) bool { + if withinExact(root, path) { + return true + } + if !within(root, path) { + return false + } + depth := len(strings.Split(filepath.Clean(root), string(filepath.Separator))) + parts := strings.Split(filepath.Clean(path), string(filepath.Separator)) + if len(parts) < depth { + return false + } + prefix := strings.Join(parts[:depth], string(filepath.Separator)) + if prefix == "" { + prefix = string(filepath.Separator) + } + held, err := os.Stat(root) + if err != nil { + return false + } + spelled, err := os.Stat(prefix) + return err == nil && os.SameFile(held, spelled) +} + +func overlaps(a, b []string) bool { + for _, x := range a { + for _, y := range b { + if within(x, y) || within(y, x) { + return true + } + } + } + left, right := existing(a), existing(b) + for _, x := range left { + for _, y := range right { + if os.SameFile(x, y) { + return true + } + } + } + return false +} + +func existing(paths []string) []os.FileInfo { + out := make([]os.FileInfo, 0, len(paths)) + for _, path := range paths { + if info, err := os.Stat(path); err == nil { + out = append(out, info) + } + } + return out +} + +func covers(held, requested []string) bool { + return !slices.ContainsFunc(requested, func(path string) bool { + return !slices.ContainsFunc(held, func(root string) bool { return heldWithin(root, path) }) + }) +} + +func (c *Coordinator) validate(ctx context.Context, i Intent, paths []string) error { + current, err := c.resources(i) + if err != nil { + return err + } + if !slices.Equal(current, paths) { + return ErrInvalidIntent + } + if c.admit != nil && i.Effect != CapacityOnly { + i.Paths = current + if err := c.admit(ctx, i); err != nil { + return err + } + } + return validateExpected(ctx, i.Expected, paths) +} + +// versionProbe rehashes a waiting request's inputs only when their metadata +// moves; the check under a granted lease still reads every version in full. +type versionProbe map[string]os.FileInfo + +func (p versionProbe) check(ctx context.Context, versions map[string]ContentVersion, paths []string) error { + for path, expected := range versions { + info, statErr := os.Lstat(path) + if previous, ok := p[path]; ok && statErr == nil && os.SameFile(previous, info) && + previous.Size() == info.Size() && previous.Mode() == info.Mode() && previous.ModTime().Equal(info.ModTime()) { + continue + } + if err := validateExpected(ctx, map[string]ContentVersion{path: expected}, paths); err != nil { + return err + } + if statErr == nil { + p[path] = info + } + } + return nil +} + +// A preflight may reject stale inputs, but only the repeated check under a +// granted lease authorizes execution against the expected versions. +func validateExpected(ctx context.Context, versions map[string]ContentVersion, paths []string) error { + for path, expected := range versions { + if !filepath.IsAbs(path) { + return ErrInvalidIntent + } + canonical, err := fileutil.CanonicalWritePath(path) + if err != nil { + return err + } + if !slices.ContainsFunc(paths, func(root string) bool { return heldWithin(root, canonical) }) { + return ErrInvalidIntent + } + actual, err := ReadVersionContext(ctx, path) + if err != nil { + return err + } + if actual != expected { + return &VersionConflict{Path: path} + } + } + return nil +} diff --git a/internal/state/mutation/stale_wait_test.go b/internal/state/mutation/stale_wait_test.go new file mode 100644 index 0000000000..81fa662458 --- /dev/null +++ b/internal/state/mutation/stale_wait_test.go @@ -0,0 +1,78 @@ +package mutation + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "testing" + "time" + + "reasonix/internal/state/workspacelease" +) + +func TestStaleWaiterLeavesBeforeCurrentWriterReleases(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "shared") + holder := testCoordinator(t, root, state) + held, err := holder.Acquire(t.Context(), intent(path)) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + var notices []Notice + waiter, err := New(root, state, func(n Notice) { + notices = append(notices, n) + if n.State == "waiting" { + if err := os.WriteFile(path, []byte("writer postimage"), 0600); err != nil { + t.Fatal(err) + } + } + }) + if err != nil { + t.Fatal(err) + } + request := intent(path) + request.Expected = map[string]ContentVersion{path: {Kind: "absent"}} + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + lease, err := waiter.Acquire(ctx, request) + lease.End(NotExecuted) + if !errors.Is(err, ErrVersionConflict) { + t.Fatalf("stale waiter waited for resource release: %v", err) + } + if len(notices) != 2 || notices[0].State != "waiting" || notices[1].State != "abandoned" || notices[1].Reason != "conflict" { + t.Fatalf("stale wait was not closed as a conflict: %+v", notices) + } + if notices[0].TransactionID != notices[1].TransactionID { + t.Fatal("conflict closed another wait") + } + if err := held.CheckPaths([]string{path}); err != nil { + t.Fatalf("stale waiter disturbed the writer: %v", err) + } + records, err := waiter.records() + if err != nil || slices.ContainsFunc(records, func(r record) bool { return r.ID != held.ID }) { + t.Fatalf("stale queue entry retained: %+v %v", records, err) + } +} + +func TestAlreadyStaleRequestDoesNotEnterCompatibilityPreparation(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "shared") + if err := os.WriteFile(path, []byte("new"), 0600); err != nil { + t.Fatal(err) + } + c := testCoordinator(t, root, state) + c.compatClaim = func(string, string, string, []string) (func(), *workspacelease.ConflictError, error) { + t.Fatal("stale request entered compatibility preparation") + return nil, nil, ErrProtocol + } + request := intent(path) + request.Expected = map[string]ContentVersion{path: {Kind: "absent"}} + lease, err := c.Acquire(t.Context(), request) + lease.End(NotExecuted) + if !errors.Is(err, ErrVersionConflict) { + t.Fatalf("stale request: %v", err) + } +} diff --git a/internal/state/mutation/try_test.go b/internal/state/mutation/try_test.go new file mode 100644 index 0000000000..fedaa01028 --- /dev/null +++ b/internal/state/mutation/try_test.go @@ -0,0 +1,38 @@ +package mutation + +import ( + "errors" + "path/filepath" + "testing" +) + +func TestTryAcquireRefusesConflictsAndLeavesNoNewOccupancy(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + coordinator := testCoordinator(t, root, state) + held, err := coordinator.Acquire(t.Context(), intent(filepath.Join(root, "a"))) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + other := testCoordinator(t, root, state) + if got, err := other.TryAcquire(t.Context(), intent(root)); got != nil || !errors.Is(err, ErrBusy) { + t.Fatalf("conflict did not refuse: %v %v", got, err) + } + records, err := other.records() + if err != nil || len(records) != 1 || records[0].ID != held.ID { + t.Fatalf("failed probe retained resources: %+v %v", records, err) + } + disjoint, err := other.TryAcquire(t.Context(), intent(filepath.Join(root, "b"))) + if err != nil { + t.Fatal(err) + } + disjoint.End(NotExecuted) + child, err := coordinator.TryAcquire(held.Context(t.Context()), intent(filepath.Join(root, "a", "child"))) + if err != nil { + t.Fatal(err) + } + child.End(NotExecuted) + if err := held.CheckPaths([]string{filepath.Join(root, "a")}); err != nil { + t.Fatal("probe released its parent's lease") + } +} diff --git a/internal/state/mutation/types.go b/internal/state/mutation/types.go new file mode 100644 index 0000000000..a808aea5b7 --- /dev/null +++ b/internal/state/mutation/types.go @@ -0,0 +1,102 @@ +package mutation + +import ( + "errors" + "time" +) + +type Effect string + +const ( + ReadOnly Effect = "read_only" + NamedPaths Effect = "named_paths" + Workspace Effect = "workspace" + GitMetadata Effect = "git_metadata" + External Effect = "external" + CapacityOnly Effect = "capacity" +) + +var ( + ErrInvalidIntent = errors.New("invalid mutation intent") + ErrVersionConflict = errors.New("mutation input changed") + ErrWaitTimeout = errors.New("mutation wait exhausted") + ErrCapacityWaitTimeout = errors.New("execution capacity wait exhausted") + ErrProtocol = errors.New("unsupported mutation coordination record") + ErrCallback = errors.New("mutation coordination callback failed") + ErrBusy = errors.New("mutation resources are busy") +) + +type Intent struct { + tryOnly bool + Effect Effect + Paths []string + SessionID string + ExecutionID string + CallID string + Expected map[string]ContentVersion + Recovery bool + Capacities []Capacity +} + +type Outcome string + +const ( + NotExecuted Outcome = "not_executed" + Committed Outcome = "committed" + Partial Outcome = "partial" + Unknown Outcome = "unknown" +) + +type Notice struct { + Effect Effect + BlockedOn Effect + Capacities []Capacity + TransactionID string + SessionID string + ExecutionID string + CallID string + Paths []string + HolderSessionID string + HolderExecutionID string + HolderTransactionID string + HolderPaths []string + State string + Reason string + StartedAt time.Time + Elapsed time.Duration + Outcome Outcome +} + +type WaitError struct { + Effect Effect + Capacities []Capacity + TransactionID string + Paths []string + Cause error +} + +func (e *WaitError) Error() string { + if e.Effect == CapacityOnly { + return "execution capacity is busy (transaction " + e.TransactionID + "); reschedule the request: " + e.Cause.Error() + } + return "mutation resources are busy (transaction " + e.TransactionID + "); reschedule the operation: " + e.Cause.Error() +} +func (e *WaitError) Unwrap() error { return e.Cause } +func (e *WaitError) Is(target error) bool { + if e.Effect == CapacityOnly { + return target == ErrCapacityWaitTimeout + } + return target == ErrWaitTimeout +} +func (e *WaitError) RefusalCode() string { + if e.Effect == CapacityOnly { + return "execution.capacity_wait_timeout" + } + return "workspace.mutation_wait_timeout" +} + +type VersionConflict struct{ Path string } + +func (e *VersionConflict) Error() string { return "file changed before the mutation: " + e.Path } +func (*VersionConflict) Is(target error) bool { return target == ErrVersionConflict } +func (*VersionConflict) RefusalCode() string { return "workspace.mutation_version_conflict" } diff --git a/internal/state/mutation/version.go b/internal/state/mutation/version.go new file mode 100644 index 0000000000..31f83a8005 --- /dev/null +++ b/internal/state/mutation/version.go @@ -0,0 +1,21 @@ +package mutation + +import ( + "context" + "errors" + + "reasonix/internal/base/fileutil" +) + +type ContentVersion = fileutil.ContentVersion + +func ReadVersionContext(ctx context.Context, path string) (ContentVersion, error) { + version, err := fileutil.ReadContentVersion(ctx, path) + if errors.Is(err, fileutil.ErrContentChanged) { + return ContentVersion{}, &VersionConflict{Path: path} + } + if errors.Is(err, errors.ErrUnsupported) { + return ContentVersion{}, ErrInvalidIntent + } + return version, err +} diff --git a/internal/state/mutation/workspace_resources_test.go b/internal/state/mutation/workspace_resources_test.go new file mode 100644 index 0000000000..41c5b3069d --- /dev/null +++ b/internal/state/mutation/workspace_resources_test.go @@ -0,0 +1,113 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "sync/atomic" + "testing" +) + +func TestOpaqueWorktreeWriteCoordinatesSharedMetadataOnlyDuringMutation(t *testing.T) { + root, linked, state := t.TempDir(), t.TempDir(), t.TempDir() + common := filepath.Join(root, ".git") + owner := testCoordinator(t, root, state) + held, err := owner.Acquire(t.Context(), Intent{Effect: GitMetadata, Paths: []string{common}}) + if err != nil { + t.Fatal(err) + } + defer held.End(NotExecuted) + notices := make(chan Notice, 8) + c, err := New(linked, state, func(n Notice) { notices <- n }, Options{WorkspaceResources: []string{common}}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + lease, err := c.Acquire(ctx, Intent{Effect: Workspace}) + if lease != nil { + lease.End(NotExecuted) + } + done <- err + }() + start := awaitCapacityNotice(t, notices, "waiting") + if start.HolderTransactionID != held.ID { + t.Fatal("opaque write did not wait for shared metadata") + } + other, err := owner.Acquire(t.Context(), Intent{Effect: NamedPaths, Paths: []string{filepath.Join(root, "independent.txt")}}) + if err != nil { + t.Fatal(err) + } + other.End(NotExecuted) + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + end := awaitCapacityNotice(t, notices, "abandoned") + if end.TransactionID != start.TransactionID { + t.Fatal("unclosed opaque metadata wait") + } +} + +func TestContinuationRepeatsOpaqueResourceAdmission(t *testing.T) { + var changed atomic.Bool + c, err := New(t.TempDir(), t.TempDir(), nil, Options{AdmissionCheck: func(_ context.Context, i Intent) error { + if i.Effect == Workspace && changed.Load() { + return ErrInvalidIntent + } + return nil + }}) + if err != nil { + t.Fatal(err) + } + lease, err := c.Acquire(t.Context(), Intent{Effect: Workspace}) + if err != nil { + t.Fatal(err) + } + continuation, err := lease.PrepareContinuation() + if err != nil { + t.Fatal(err) + } + lease.End(NotExecuted) + changed.Store(true) + if next, err := continuation.Acquire(t.Context()); !errors.Is(err, ErrInvalidIntent) { + if next != nil { + next.End(NotExecuted) + } + t.Fatalf("background admission forgot opaque resource identity: %v", err) + } +} + +func TestContextAdmissionCancellationReleasesAllResources(t *testing.T) { + entered := make(chan struct{}) + root, state := t.TempDir(), t.TempDir() + c, err := New(root, state, nil, Options{AdmissionCheck: func(ctx context.Context, _ Intent) error { + close(entered) + <-ctx.Done() + return ctx.Err() + }}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + lease, err := c.Acquire(ctx, Intent{Effect: Workspace}) + lease.End(NotExecuted) + done <- err + }() + <-entered + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("cancelled admission: %v", err) + } + other := testCoordinator(t, root, state) + lease, err := other.Acquire(t.Context(), Intent{Effect: Workspace}) + if err != nil { + t.Fatal(err) + } + lease.End(NotExecuted) +} diff --git a/internal/state/workspacelease/claim_attempt.go b/internal/state/workspacelease/claim_attempt.go new file mode 100644 index 0000000000..22e87acd53 --- /dev/null +++ b/internal/state/workspacelease/claim_attempt.go @@ -0,0 +1,45 @@ +package workspacelease + +import ( + "errors" + "sync" +) + +var ErrClaimBusy = errors.New("workspace claim cannot be acquired immediately") +var ErrInvalidClaim = errors.New("workspace claim requires explicit resolvable paths") + +// TryClaimPaths makes one non-blocking, independent compatibility claim. +// A successful caller owns the idempotent release, without an Owner run lifetime. +func TryClaimPaths(root, lockDir, sessionID string, paths []string) (func(), *ConflictError, error) { + return claimAttempt(root, lockDir, sessionID, paths, true) +} + +// ProbeClaimPaths checks current compatibility contention without retaining claims. +func ProbeClaimPaths(root, lockDir string, paths []string) (*ConflictError, error) { + _, conflict, err := claimAttempt(root, lockDir, "", paths, false) + return conflict, err +} + +func claimAttempt(root, lockDir, sessionID string, paths []string, publish bool) (func(), *ConflictError, error) { + if len(paths) == 0 { + return nil, nil, ErrInvalidClaim + } + o, err := New(root, lockDir, nil) + if err != nil { + return nil, nil, err + } + o.scope.sessionID = func() string { return sessionID } + normalized := o.normalizePaths(paths) + if len(normalized) != len(paths) { + return nil, nil, ErrInvalidClaim + } + release, conflict, err := o.tryPathSet(normalized, false, publish) + if errors.Is(err, errHeld) { + return nil, conflict, ErrClaimBusy + } + if err != nil || release == nil { + return nil, conflict, err + } + var once sync.Once + return func() { once.Do(release) }, nil, nil +} diff --git a/internal/state/workspacelease/claim_attempt_test.go b/internal/state/workspacelease/claim_attempt_test.go new file mode 100644 index 0000000000..b54f4282bf --- /dev/null +++ b/internal/state/workspacelease/claim_attempt_test.go @@ -0,0 +1,60 @@ +package workspacelease + +import ( + "errors" + "path/filepath" + "testing" +) + +func TestClaimAttemptAndProbeDoNotRetainPartialClaims(t *testing.T) { + root, locks := t.TempDir(), t.TempDir() + paths := []string{filepath.Join(root, "a")} + if _, err := ProbeClaimPaths(root, locks, paths); err != nil { + t.Fatal(err) + } + first, _, err := TryClaimPaths(root, locks, "first", paths) + if err != nil { + t.Fatal(err) + } + defer first() + for _, probe := range []bool{false, true} { + var conflict *ConflictError + if probe { + conflict, err = ProbeClaimPaths(root, locks, paths) + } else { + _, conflict, err = TryClaimPaths(root, locks, "second", paths) + } + if !errors.Is(err, ErrClaimBusy) || conflict == nil || conflict.SessionID != "first" { + t.Fatalf("missing immediate conflict: %+v %v", conflict, err) + } + } + first() + second, _, err := TryClaimPaths(root, locks, "second", paths) + if err != nil { + t.Fatal(err) + } + defer second() + first() + if conflict, err := ProbeClaimPaths(root, locks, paths); !errors.Is(err, ErrClaimBusy) || conflict.SessionID != "second" { + t.Fatalf("old release cleared a new claim: %+v %v", conflict, err) + } + if _, _, err := TryClaimPaths(root, locks, "invalid", nil); !errors.Is(err, ErrInvalidClaim) { + t.Fatal(err) + } +} + +func TestClaimAttemptHonorsLegacyWholeWorkspaceLock(t *testing.T) { + root, locks := t.TempDir(), t.TempDir() + o, err := New(root, locks, nil) + if err != nil { + t.Fatal(err) + } + release, err := tryLockFile(o.lockPath) + if err != nil { + t.Fatal(err) + } + defer release() + if _, _, err := TryClaimPaths(root, locks, "new", []string{filepath.Join(root, "file")}); !errors.Is(err, ErrClaimBusy) { + t.Fatalf("bypassed old exclusive lock: %v", err) + } +} diff --git a/internal/state/workspacelease/claim_panic_test.go b/internal/state/workspacelease/claim_panic_test.go new file mode 100644 index 0000000000..7bac287f99 --- /dev/null +++ b/internal/state/workspacelease/claim_panic_test.go @@ -0,0 +1,46 @@ +package workspacelease + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "reasonix/internal/base/fileutil" +) + +func TestCompatibilityClaimPanicBeforeHandoffReleasesItsNewLocks(t *testing.T) { + root, state := t.TempDir(), t.TempDir() + path := filepath.Join(root, "a") + before := fileutil.CrashPoint + fileutil.CrashPoint = func(op, target string) { + if op == "atomic-write" && strings.Contains(target, ".claims") { + panic("claim write failed before ownership handoff") + } + } + func() { + defer func() { + if recover() == nil { + t.Error("fixture did not fail at claim publication") + } + }() + _, _, _ = TryClaimPaths(root, state, "first", []string{path}) + }() + fileutil.CrashPoint = before + t.Cleanup(func() { fileutil.CrashPoint = before }) + claimDirs, err := filepath.Glob(filepath.Join(state, "*.claims")) + if err != nil || len(claimDirs) == 0 { + t.Fatalf("fixture did not reach claim publication: %v", err) + } + for _, dir := range claimDirs { + entries, err := os.ReadDir(dir) + if err != nil || len(entries) != 0 { + t.Fatalf("failed handoff retained unpublished claim resources: %v %v", entries, err) + } + } + release, _, err := TryClaimPaths(root, state, "second", []string{path}) + if err != nil { + t.Fatalf("failed handoff retained physical compatibility ownership: %v", err) + } + release() +} diff --git a/internal/state/workspacelease/paths.go b/internal/state/workspacelease/paths.go index 6ab2885869..b5f72c7a02 100644 --- a/internal/state/workspacelease/paths.go +++ b/internal/state/workspacelease/paths.go @@ -151,6 +151,10 @@ func (o *Owner) acquirePaths(ctx context.Context, paths []string, held bool) (fu } func (o *Owner) tryPaths(paths []string, held bool) (func(), *ConflictError, error) { + return o.tryPathSet(paths, held, true) +} + +func (o *Owner) tryPathSet(paths []string, held, publish bool) (func(), *ConflictError, error) { select { case <-o.local.token: default: @@ -224,6 +228,9 @@ func (o *Owner) tryPaths(paths []string, held bool) (func(), *ConflictError, err return nil, o.conflict(record.Holder, record.SessionID, record.Paths, paths), errHeld } } + if !publish { + return nil, nil, nil + } if !held { var id [16]byte if _, err := rand.Read(id[:]); err != nil { @@ -239,18 +246,24 @@ func (o *Owner) tryPaths(paths []string, held bool) (func(), *ConflictError, err return nil, nil, err } } - if err := o.writeClaim(path, paths); err != nil { + defer func() { if release != nil { + _ = os.Remove(path) release() + _ = os.Remove(path + ".lock") } + }() + if err := o.writeClaim(path, paths); err != nil { return nil, nil, err } if held { return nil, nil, nil } sharedRelease := shared + claimRelease := release + release = nil shared = nil - return func() { o.releaseClaim(path, release); sharedRelease() }, nil, nil + return func() { o.releaseClaim(path, claimRelease); sharedRelease() }, nil, nil } func (o *Owner) releaseClaim(path string, release func()) {