mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
feat(mutation): admit explicitly opaque write effects
Problem: Directory-scoped coordination cannot exclude writers that address a source file through a hardlink outside that directory. Root cause: Path containment and named inode identity do not represent an unresolved write effect; reinterpreting old directory records would silently change their resource promise and nested admission could reuse a weaker hold. Fix: Publish explicit version-3 opaque effects in the existing registry, coordinate them symmetrically against all path-bearing transactions, and require effect coverage when reusing a parent lease. Preserve ordinary named parallelism, model capacity, previous directory semantics and final OS-locked grant authority. No runtime or default requests the new effect here. Verification: Full 239-package root inventory and independent builtin/boot commands exit 0; mutation race 82 passes/zero skips; preceding-reader binary preserves live/dead new records; weakened-owner control fails eight cases; selected wait-snapshot composition has 36 race passes/zero skips. Root vet, full lint, changed-file checks and formatting pass at unchanged baseline 2664. Windows/Linux amd64 compile only; local execution is macOS.
This commit is contained in:
1 parent
9fce63aa5a
commit
25169a45f1
11 files changed
+580
-3
No files matched your search
@@ -0,0 +1,125 @@
|
||||
---
|
||||
owner: @SivanCola
|
||||
backup: @esengine
|
||||
status: active
|
||||
reviewed: 2026-10-25
|
||||
---
|
||||
|
||||
# Opaque write admission / 未知范围写入准入
|
||||
|
||||
## Scope and cause / 范围与根因
|
||||
|
||||
A directory reservation cannot enumerate hardlink aliases outside its tree.
|
||||
The actual filesystem probe holds a Workspace lease, then admits a named write
|
||||
through an external hardlink and changes the held source.
|
||||
|
||||
Named-file aliases and direct descendants are already excluded. This is a documented primitive
|
||||
limitation, not an observed production incident; shared runtime activation is
|
||||
still separate.
|
||||
|
||||
`OpaqueWrite` explicitly requests conservative exclusion across all path-bearing
|
||||
transactions in the same trusted registry. Named disjoint writes still proceed
|
||||
concurrently. Capacity-only/model requests do not participate in that exclusion.
|
||||
No tool, host adapter, permission, default or prompt definition changes here.
|
||||
|
||||
目录范围不能枚举树外硬链接。真实文件探针证明旧 Workspace 租约不能阻止通过该
|
||||
别名的具名写入;具名文件别名和目录内直接路径已有互斥。本次明确新增
|
||||
`OpaqueWrite`,不悄悄扩大旧目录协议含义,也不启用运行时新策略。
|
||||
|
||||
## Source, sink and authority / 输入、副作用与权威
|
||||
|
||||
- Hosts select `Intent.Effect` and the trusted state directory. This package
|
||||
does not parse tool arguments, approve commands or authorize filesystem access.
|
||||
- Opaque records publish version 3; only this effect is supported in version 3.
|
||||
Versions 1 and 2 retain their existing path/capacity semantics and encoding.
|
||||
The shared `mutations-v1` directory remains the admission authority; a separate
|
||||
directory would let old named writers miss the new owner entirely.
|
||||
- One symmetric resource relation makes opaque owners and older opaque waiters
|
||||
conflict with every path-bearing request, including other roots and Git paths.
|
||||
- Queue order, complete-set admission, live OS ownership, compatibility claims
|
||||
and the final fresh locked check remain in the common owner. No new singleton
|
||||
execution lock is added to every named writer.
|
||||
- The paths retained on an opaque lease still constrain nested calls, expected
|
||||
content versions and compatibility claims. Global exclusion is not global
|
||||
filesystem permission. A directory parent cannot satisfy an opaque child.
|
||||
- Continuations retain the opaque effect and acquire a new transaction. Cancel
|
||||
and process death use the existing identified cleanup, not forced lock theft.
|
||||
|
||||
影响类型由宿主选择,权限检查仍属于宿主。新版本记录显式表达全注册表互斥,旧版本
|
||||
保持原义。嵌套调用同时检查路径、额度和影响覆盖,不能靠父目录将普通租约升级为
|
||||
全域租约;全域互斥也不扩大嵌套路径授权。
|
||||
|
||||
## Compatibility / 兼容性
|
||||
|
||||
| Input or participant | New coordinator | Preceding coordinator |
|
||||
| --- | --- | --- |
|
||||
| Known version 1/2 records | Existing semantics | Existing semantics |
|
||||
| Version 3 opaque owner, live or dead | Interpret/reclaim when understood | Refuse and preserve bytes |
|
||||
| Version 1/2 opaque effect | Refuse and preserve | Unsupported semantics; do not publish |
|
||||
| Version 3 nonopaque effect or missing paths | Refuse and preserve | Refuse and preserve |
|
||||
| Unknown version/field | Refuse and preserve | Version refusal; field behavior is version-specific |
|
||||
| Legacy workspacelease writer | Existing workspace claim coverage | No registry-wide opaque guarantee |
|
||||
|
||||
The preceding accepted coordinator is compiled and executed against live/dead
|
||||
version-3 records, with real legacy cleanup as a control. No new normal runtime
|
||||
publishes these records yet.
|
||||
|
||||
While one exists, preceding coordinators refuse all new admission instead of
|
||||
trying to interpret global semantics. That is an intentional availability
|
||||
boundary; an implementation that understands version 3
|
||||
can reclaim a known dead owner. Do not delete unknown records to bypass it.
|
||||
|
||||
已接受的前版通过真实编译二进制验证:新记录持有者存活或死亡时都保留字节并拒绝
|
||||
准入,旧记录清理控制正常执行。旧程序遇新协议会阻塞,必须由理解该协议的实现
|
||||
处理;不能通过手删记录制造并发。旧 workspacelease 写入者仍只有原工作区域保障。
|
||||
|
||||
## Evidence / 验证
|
||||
|
||||
- Real hardlink aliases in both acquisition orders, with an actual write after
|
||||
release proving the alias is effective; unrelated-root cross-process owners
|
||||
exclude each other and recover after the child is killed and joined.
|
||||
- Disjoint named writers proceed; active opaque ownership and queued opaque
|
||||
writes do not consume unrelated model capacity. Cancellation closes the wait
|
||||
notice and removes the new contribution without releasing the active owner.
|
||||
- An older opaque waiter prevents later named bypass. Nested directory upgrades
|
||||
refuse; outside nested paths refuse even under opaque ownership. Continuation
|
||||
retains exclusion after the parent ends.
|
||||
- Versioned publication with and without capacities, legacy directory semantics,
|
||||
unknown versions/fields and invalid resource shapes are covered separately.
|
||||
- Deliberately restoring path-only conflict and path-only nested coverage
|
||||
compiles and breaks the new invariants. This is a weakened-control test of a
|
||||
new effect, not proof that the preceding API already promised that effect.
|
||||
|
||||
真实文件、子进程、队列、取消、嵌套、续接和版本兼容分别验证。故意削弱新规则的
|
||||
对照用于证明测试有效,不能冒充旧版已实现新接口却违反承诺的缺陷复现。
|
||||
|
||||
## Limits and next integration / 边界与后续接入
|
||||
|
||||
This is cooperative exclusion inside one trusted state namespace. Other state
|
||||
directories, arbitrary processes and external services remain outside it.
|
||||
The registry is not authenticated or bounded by this change. Existing Workspace
|
||||
and directory-valued NamedPaths leases retain their documented alias limitation.
|
||||
|
||||
Before runtime activation, map unresolved tool/hook effects explicitly, authorize
|
||||
first and keep the lease only during physical mutation.
|
||||
|
||||
Lifetime completion, unknown-scope recovery, strict per-actor reads/CAS and evidence remain separate.
|
||||
Never hold it through thinking, queuing or user waits. This unit does not close
|
||||
the future shell/hook completion API or automatically select isolation.
|
||||
|
||||
Applying this global effect to every worktree command would serialize independent
|
||||
tasks. A private execution adapter needs demonstrated write-root, alias, backend
|
||||
and shared-Git boundaries before choosing narrower coordination.
|
||||
|
||||
A changed cwd or initially distinct inodes does not prove isolation. No worktree availability,
|
||||
complete sandbox, Windows runtime or final performance claim is made here.
|
||||
|
||||
同一可信注册表以外的进程和外部副作用不受约束。后续必须先权限检查,再按真实影响
|
||||
接入并限定物理执行生命周期;模型思考、任务排队和等待用户不应持有该租约。
|
||||
不能对所有 worktree 命令简单套全域锁后声称隔离任务仍可并行。私有执行域需要实际
|
||||
后端、授权根、别名和共享 Git 边界证据。本次不是完整沙箱或最终性能验收。
|
||||
|
||||
This is the author's security assessment, not independent approval or a
|
||||
disposition of the original large PR's CodeQL findings.
|
||||
|
||||
本说明为作者评估,不替代独立安全审查,也不批量判定原大 PR 的告警。
|
||||
@@ -41,6 +41,10 @@ func coversCapacities(held, requested []Capacity) bool {
|
||||
}
|
||||
|
||||
func validRecordResources(r record) bool {
|
||||
if r.Version == 3 {
|
||||
_, err := normalizedCapacities(r.Capacities)
|
||||
return r.Effect == OpaqueWrite && len(r.Paths) > 0 && filepath.IsAbs(r.Workspace) && err == nil
|
||||
}
|
||||
if r.Version == 1 {
|
||||
// The original path-only protocol permitted an omitted effect.
|
||||
knownEffect := r.Effect == "" || r.Effect == NamedPaths || r.Effect == Workspace || r.Effect == GitMetadata
|
||||
|
||||
@@ -99,7 +99,8 @@ func (c *Coordinator) Acquire(ctx context.Context, intent Intent) (acquired *Lea
|
||||
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) {
|
||||
if parent.closed.Load() || !coversEffect(parent.record.Effect, intent.Effect) ||
|
||||
!covers(parent.record.Paths, paths) || !coversCapacities(parent.record.Capacities, capacities) {
|
||||
return nil, ErrInvalidIntent
|
||||
}
|
||||
if err := c.validate(ctx, intent, paths); err != nil {
|
||||
@@ -118,6 +119,9 @@ func (c *Coordinator) Acquire(ctx context.Context, intent Intent) (acquired *Lea
|
||||
if len(capacities) != 0 {
|
||||
l.record.Version = 2
|
||||
}
|
||||
if intent.Effect == OpaqueWrite {
|
||||
l.record.Version = 3
|
||||
}
|
||||
l.record.ID = l.ID
|
||||
defer func() {
|
||||
if errors.Is(resultErr, context.DeadlineExceeded) {
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
package mutation
|
||||
|
||||
// Opaque exclusion spans the trusted registry, including tree-external aliases.
|
||||
// Its paths still bound nested admission and legacy workspace compatibility.
|
||||
func recordsOverlap(a, b record) bool {
|
||||
if len(a.Paths) == 0 || len(b.Paths) == 0 {
|
||||
return false
|
||||
}
|
||||
return a.Effect == OpaqueWrite || b.Effect == OpaqueWrite || overlaps(a.Paths, b.Paths)
|
||||
}
|
||||
|
||||
func coversEffect(held, requested Effect) bool {
|
||||
return requested != OpaqueWrite || held == OpaqueWrite
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package mutation
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestOpaqueProcessHelper(t *testing.T) {
|
||||
if os.Getenv("REASONIX_OPAQUE_PROCESS") != "1" {
|
||||
return
|
||||
}
|
||||
c, err := New(os.Getenv("REASONIX_OPAQUE_ROOT"), os.Getenv("REASONIX_OPAQUE_STATE"), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request := opaqueIntent()
|
||||
if os.Getenv("REASONIX_OPAQUE_NAMED") == "1" {
|
||||
request = intent("file")
|
||||
}
|
||||
lease, err := c.Acquire(t.Context(), request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer lease.End(NotExecuted)
|
||||
fmt.Println("opaque-ready")
|
||||
_, _ = bufio.NewReader(os.Stdin).ReadString('\n')
|
||||
}
|
||||
|
||||
func TestOpaqueWriteCrossProcessAndDeath(t *testing.T) {
|
||||
for _, namedChild := range []string{"0", "1"} {
|
||||
t.Run("named_child_"+namedChild, func(t *testing.T) {
|
||||
root, extra, state := t.TempDir(), t.TempDir(), t.TempDir()
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestOpaqueProcessHelper$")
|
||||
cmd.Env = append(os.Environ(), "REASONIX_OPAQUE_PROCESS=1", "REASONIX_OPAQUE_ROOT="+root,
|
||||
"REASONIX_OPAQUE_STATE="+state, "REASONIX_OPAQUE_NAMED="+namedChild)
|
||||
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 != "opaque-ready\n" {
|
||||
t.Fatal(line)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal(ctx.Err())
|
||||
}
|
||||
c := testCoordinator(t, extra, state)
|
||||
request := intent(filepath.Join(extra, "unrelated"))
|
||||
if namedChild == "1" {
|
||||
request = opaqueIntent()
|
||||
}
|
||||
if lease, err := c.TryAcquire(ctx, request); !errors.Is(err, ErrBusy) {
|
||||
if lease != nil {
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
t.Fatalf("cross-process global exclusion absent: %v", err)
|
||||
}
|
||||
if err := cmd.Process.Kill(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = cmd.Wait()
|
||||
lease, err := c.Acquire(ctx, request)
|
||||
if err != nil {
|
||||
t.Fatalf("dead owner not reclaimed: %v", err)
|
||||
}
|
||||
lease.End(NotExecuted)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package mutation
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/base/filelock"
|
||||
)
|
||||
|
||||
func TestOpaqueWritePublishesExplicitVersionAndPreservesLegacyDirectoryMeaning(t *testing.T) {
|
||||
root, state := t.TempDir(), t.TempDir()
|
||||
c := testCoordinator(t, root, state)
|
||||
for _, pool := range []bool{false, true} {
|
||||
request := opaqueIntent()
|
||||
if pool {
|
||||
request.Capacities = []Capacity{capacity("model")}
|
||||
}
|
||||
lease, err := c.Acquire(t.Context(), request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, err := os.ReadFile(c.recordPath(lease.ID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var r record
|
||||
if err := json.Unmarshal(raw, &r); err != nil || r.Version != 3 || r.Effect != OpaqueWrite {
|
||||
t.Fatalf("opaque semantics published under an old version: %s %v", raw, err)
|
||||
}
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
legacy, err := c.Acquire(t.Context(), Intent{Effect: Workspace})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer legacy.End(NotExecuted)
|
||||
other := testCoordinator(t, t.TempDir(), state)
|
||||
named, err := other.TryAcquire(t.Context(), intent("unrelated"))
|
||||
if err != nil {
|
||||
t.Fatalf("legacy directory silently became global: %v", err)
|
||||
}
|
||||
named.End(NotExecuted)
|
||||
}
|
||||
|
||||
func TestOpaqueUnsupportedProtocolPreservesLiveAndDeadRecords(t *testing.T) {
|
||||
for _, scenario := range []string{"old_version", "future_version", "other_effect", "no_paths", "unknown_field"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
c := testCoordinator(t, t.TempDir(), t.TempDir())
|
||||
r := record{Version: 3, ID: "opaque", Order: 1, Active: true, Effect: OpaqueWrite,
|
||||
Workspace: c.root, Paths: []string{c.root}}
|
||||
switch scenario {
|
||||
case "old_version":
|
||||
r.Version = 1
|
||||
case "future_version":
|
||||
r.Version = 4
|
||||
case "other_effect":
|
||||
r.Effect = NamedPaths
|
||||
case "no_paths":
|
||||
r.Paths = nil
|
||||
}
|
||||
raw, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if scenario == "unknown_field" {
|
||||
raw = append(raw[:len(raw)-1], []byte(`,"future":true}`)...)
|
||||
}
|
||||
if err := os.WriteFile(c.recordPath(r.ID), raw, 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release, err := filelock.TryAcquire(c.livePath(r.ID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer release()
|
||||
for _, phase := range []string{"live", "dead"} {
|
||||
if phase == "dead" {
|
||||
release()
|
||||
}
|
||||
lease, err := c.TryAcquire(t.Context(), intent(filepath.Join(c.root, "new")))
|
||||
if lease != nil {
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
if lease != nil || !errors.Is(err, ErrProtocol) {
|
||||
t.Errorf("%s invalid record admitted mutation: %v", phase, err)
|
||||
}
|
||||
actual, err := os.ReadFile(c.recordPath(r.ID))
|
||||
if err != nil || string(actual) != string(raw) {
|
||||
t.Errorf("%s invalid record changed: %q %v", phase, actual, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,236 @@
|
||||
package mutation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func opaqueIntent() Intent { return Intent{Effect: Effect("opaque_write")} }
|
||||
|
||||
func TestOpaqueWriteExcludesExternalHardlinkInBothOrders(t *testing.T) {
|
||||
for _, opaqueFirst := range []bool{true, false} {
|
||||
t.Run(map[bool]string{true: "opaque_first", false: "named_first"}[opaqueFirst], func(t *testing.T) {
|
||||
root, extra, state := t.TempDir(), t.TempDir(), t.TempDir()
|
||||
source, alias := filepath.Join(root, "source"), filepath.Join(extra, "alias")
|
||||
if err := os.WriteFile(source, []byte("original"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Link(source, alias); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a, b := testCoordinator(t, root, state), testCoordinator(t, extra, state)
|
||||
first, second := opaqueIntent(), intent(alias)
|
||||
if !opaqueFirst {
|
||||
a, b, first, second = b, a, second, first
|
||||
}
|
||||
held, err := a.Acquire(t.Context(), first)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.End(NotExecuted)
|
||||
blocked, err := b.TryAcquire(t.Context(), second)
|
||||
if blocked != nil {
|
||||
blocked.End(NotExecuted)
|
||||
}
|
||||
if !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("cross-root alias admitted: %v", err)
|
||||
}
|
||||
held.End(NotExecuted)
|
||||
next, err := b.TryAcquire(t.Context(), second)
|
||||
if err != nil {
|
||||
t.Fatalf("ended opaque owner still blocks: %v", err)
|
||||
}
|
||||
defer next.End(Committed)
|
||||
if err := os.WriteFile(alias, []byte("changed"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
actual, err := os.ReadFile(source)
|
||||
if err != nil || string(actual) != "changed" {
|
||||
t.Fatalf("alias control did not mutate source: %q %v", actual, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpaqueWriteDoesNotSerializeNamedOrModelWork(t *testing.T) {
|
||||
root, extra, state := t.TempDir(), t.TempDir(), t.TempDir()
|
||||
a, b := testCoordinator(t, root, state), testCoordinator(t, extra, state)
|
||||
held, err := a.Acquire(t.Context(), intent("first"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.End(NotExecuted)
|
||||
disjoint, err := b.TryAcquire(t.Context(), intent("second"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
disjoint.End(NotExecuted)
|
||||
held.End(NotExecuted)
|
||||
opaque, err := a.Acquire(t.Context(), opaqueIntent())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer opaque.End(NotExecuted)
|
||||
model, err := b.TryAcquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("model")}})
|
||||
if err != nil {
|
||||
t.Fatalf("opaque file owner consumed unrelated model capacity: %v", err)
|
||||
}
|
||||
model.End(NotExecuted)
|
||||
for _, effect := range []Effect{ReadOnly, External} {
|
||||
lease, err := b.TryAcquire(t.Context(), Intent{Effect: effect})
|
||||
if err != nil {
|
||||
t.Fatalf("non-file admission: %v", err)
|
||||
}
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpaqueWriteNestedCoverageAndContinuation(t *testing.T) {
|
||||
c := testCoordinator(t, t.TempDir(), t.TempDir())
|
||||
for _, parent := range []Intent{{Effect: Workspace}, intent(c.root), opaqueIntent()} {
|
||||
held, err := c.Acquire(t.Context(), parent)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := held.Context(t.Context())
|
||||
child, err := c.TryAcquire(ctx, opaqueIntent())
|
||||
if parent.Effect != opaqueIntent().Effect {
|
||||
if child != nil || !errors.Is(err, ErrInvalidIntent) {
|
||||
t.Fatalf("directory scope upgraded to opaque: %v", err)
|
||||
}
|
||||
} else {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
child.End(NotExecuted)
|
||||
}
|
||||
if _, err := c.TryAcquire(ctx, intent(filepath.Join(t.TempDir(), "outside"))); !errors.Is(err, ErrInvalidIntent) {
|
||||
t.Fatalf("global exclusion expanded nested authority: %v", err)
|
||||
}
|
||||
held.End(NotExecuted)
|
||||
}
|
||||
held, err := c.Acquire(t.Context(), opaqueIntent())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.End(NotExecuted)
|
||||
continuation, err := held.PrepareContinuation()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held.End(NotExecuted)
|
||||
next, err := continuation.Acquire(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer next.End(NotExecuted)
|
||||
other := testCoordinator(t, t.TempDir(), c.compatDir)
|
||||
if lease, err := other.TryAcquire(t.Context(), intent("outside")); !errors.Is(err, ErrBusy) {
|
||||
if lease != nil {
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
t.Fatalf("continuation lost opaque exclusion: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpaqueWriteQueueCancellationDoesNotConsumeModelCapacity(t *testing.T) {
|
||||
root, state := t.TempDir(), t.TempDir()
|
||||
held, err := testCoordinator(t, root, state).Acquire(t.Context(), intent("held"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.End(NotExecuted)
|
||||
notices := make(chan Notice, 8)
|
||||
c, err := New(t.TempDir(), state, func(n Notice) { notices <- n })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
joined := make(chan struct{})
|
||||
t.Cleanup(func() { cancel(); <-joined })
|
||||
request := opaqueIntent()
|
||||
request.Capacities = []Capacity{capacity("model")}
|
||||
go func() {
|
||||
defer close(joined)
|
||||
lease, err := c.Acquire(ctx, request)
|
||||
if lease != nil {
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
done <- err
|
||||
}()
|
||||
awaitCapacityNotice(t, notices, "waiting")
|
||||
model, err := c.TryAcquire(t.Context(), Intent{Effect: CapacityOnly, Capacities: []Capacity{capacity("model")}})
|
||||
if err != nil {
|
||||
t.Fatalf("file waiter reserved model capacity: %v", err)
|
||||
}
|
||||
model.End(NotExecuted)
|
||||
cancel()
|
||||
if err := <-done; !errors.Is(err, context.Canceled) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := awaitCapacityNotice(t, notices, "abandoned"); n.Reason != "cancelled" {
|
||||
t.Fatal(n)
|
||||
}
|
||||
next, err := c.TryAcquire(t.Context(), intent("disjoint"))
|
||||
if err != nil {
|
||||
t.Fatalf("cancelled opaque waiter retained global exclusion: %v", err)
|
||||
}
|
||||
next.End(NotExecuted)
|
||||
}
|
||||
|
||||
func TestOpaqueWriteQueuePreventsNamedBypass(t *testing.T) {
|
||||
root, state := t.TempDir(), t.TempDir()
|
||||
c := testCoordinator(t, root, state)
|
||||
held, err := c.Acquire(t.Context(), intent("held"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.End(NotExecuted)
|
||||
notices := make(chan Notice, 8)
|
||||
waiter, err := New(t.TempDir(), state, func(n Notice) { notices <- n })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
joined, release := make(chan struct{}), make(chan struct{})
|
||||
t.Cleanup(func() { cancel(); <-joined })
|
||||
go func() {
|
||||
defer close(joined)
|
||||
lease, err := waiter.Acquire(ctx, opaqueIntent())
|
||||
done <- err
|
||||
if lease != nil {
|
||||
defer lease.End(NotExecuted)
|
||||
select {
|
||||
case <-release:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
}()
|
||||
awaitCapacityNotice(t, notices, "waiting")
|
||||
if lease, err := c.TryAcquire(t.Context(), intent("unrelated")); !errors.Is(err, ErrBusy) {
|
||||
if lease != nil {
|
||||
lease.End(NotExecuted)
|
||||
}
|
||||
t.Fatalf("later named request bypassed opaque queue head: %v", err)
|
||||
}
|
||||
held.End(NotExecuted)
|
||||
if err := <-done; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
awaitCapacityNotice(t, notices, "acquired")
|
||||
close(release)
|
||||
<-joined
|
||||
next, err := c.TryAcquire(t.Context(), intent("unrelated"))
|
||||
if err != nil {
|
||||
t.Fatalf("released opaque queue owner still blocks: %v", err)
|
||||
}
|
||||
next.End(NotExecuted)
|
||||
}
|
||||
@@ -209,7 +209,7 @@ func (c *Coordinator) recordsBlocker(r record, records []record) (*blocker, erro
|
||||
|
||||
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) {
|
||||
if other.ID != r.ID && (other.Active || other.Order < r.Order) && recordsOverlap(r, other) {
|
||||
return &other
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,7 +25,7 @@ func (c *Coordinator) resources(i Intent) ([]string, error) {
|
||||
return nil, ErrInvalidIntent
|
||||
}
|
||||
return nil, nil
|
||||
case Workspace:
|
||||
case Workspace, OpaqueWrite:
|
||||
paths = append([]string{c.root}, paths...)
|
||||
paths = append(paths, c.workspaceResources...)
|
||||
case NamedPaths, GitMetadata:
|
||||
|
||||
@@ -14,6 +14,7 @@ const (
|
||||
GitMetadata Effect = "git_metadata"
|
||||
External Effect = "external"
|
||||
CapacityOnly Effect = "capacity"
|
||||
OpaqueWrite Effect = "opaque_write"
|
||||
)
|
||||
|
||||
var (
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
## 变更
|
||||
|
||||
- 为范围未知的写入增加显式事务原语,跨根路径及硬链接别名保守互斥,同时保留
|
||||
不相交具名写入与模型额度独立;尚未接入运行时或改变默认策略。Refs #14025
|
||||
- Add explicit opaque-write admission across cooperating registry users while
|
||||
preserving disjoint named writes and independent model capacity. Runtime and
|
||||
default activation remain separate. Refs #14025
|
||||
Reference in new issue
Block a user