Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/deleted-root-lsp.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- Language servers that standalone AFT started for a project now shut down as soon as that project folder is deleted, instead of running until the hour-long idle timeout. Thanks to @chrisolszewski for reporting and diagnosing this (#631).
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/disk-limit-health-cpu.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- The background check that keeps index views under their disk limit no longer re-measures unchanged folders on every pass (about 94% fewer file checks), and the health summary does less work per tick.
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/inspect-lease-flake.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- `aft_inspect` no longer reports a writer-lease timeout when no other process holds its cache but preparing the cache was slow.
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/semantic-fill-io.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Semantic view fills reuse resident chunk vectors and unchanged files while avoiding same-epoch touch write transactions, reducing disk I/O across linked checkouts.
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/swift-syntax-unknown.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- A new Swift file is no longer reported as having a syntax error when the Swift compiler could not finish checking it; AFT reports the check as unknown instead.
1 change: 1 addition & 0 deletions .alfonso/release-notes/next/task-descriptors.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- Finished and recovered background commands no longer keep a folder handle open each, so a long command history can no longer exhaust the daemon's open-file limit.
2 changes: 1 addition & 1 deletion benchmarks/aft-search/run_search_quality.py
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,7 @@ def run(args: argparse.Namespace) -> int:
if args.self_test:
command([sys.executable, str(HERE / "search_quality.py"), "--self-test"])
command(
[sys.executable, "-m", "unittest", "-v", "test_run_real_query.py", "test_run_concept_recall.py", "test_search_quality.py", "test_harness_integrity.py"],
[sys.executable, "-m", "unittest", "-v", "test_run_real_query.py", "test_run_concept_recall.py", "test_search_quality.py", "test_harness_integrity.py", "test_merge_queue_search.py"],
cwd=HERE,
)
return 0
Expand Down
8 changes: 8 additions & 0 deletions benchmarks/aft-search/slice-descriptors/land-batch-1020k.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
{
"targeted_mechanism": "none",
"kind": "harness",
"slice_class": "non_ranking",
"fixtures": [
"harness-goldens"
]
}
158 changes: 158 additions & 0 deletions benchmarks/aft-search/test_merge_queue_search.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
from __future__ import annotations

import importlib.util
from pathlib import Path
import unittest


ROOT = Path(__file__).resolve().parents[2]
SPEC = importlib.util.spec_from_file_location(
"merge_queue_search", ROOT / "scripts/merge-queue-search.py"
)
assert SPEC is not None and SPEC.loader is not None
merge_queue_search = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(merge_queue_search)


class MemberPrRetryTests(unittest.TestCase):
def setUp(self) -> None:
self.pr = {
"number": 23,
"head": {"sha": "a" * 40, "ref": "feature"},
"base": {"ref": "main", "repo": {"full_name": "cortexkit/aft"}},
}

def test_empty_lookup_retries_until_association_appears(self) -> None:
responses = iter(([], [self.pr]))
calls: list[tuple[str, str]] = []
delays: list[float] = []

def lookup(repository: str, head: str) -> list[dict]:
calls.append((repository, head))
return next(responses)

result = merge_queue_search.member_pr(
"cortexkit/aft", self.pr["head"]["sha"], lookup=lookup, sleep=delays.append
)

self.assertEqual(result, self.pr)
self.assertEqual(calls, [("cortexkit/aft", self.pr["head"]["sha"])] * 2)
self.assertEqual(delays, [5])

def test_persistent_empty_lookup_fails_after_bounded_retries(self) -> None:
calls: list[tuple[str, str]] = []
delays: list[float] = []

def lookup(repository: str, head: str) -> list[dict]:
calls.append((repository, head))
return []

with self.assertRaisesRegex(
merge_queue_search.InputFault,
r"ambiguous_queue_member:.*:matches=0:attempts=5$",
):
merge_queue_search.member_pr(
"cortexkit/aft",
self.pr["head"]["sha"],
lookup=lookup,
sleep=delays.append,
)

self.assertEqual(len(calls), 5)
self.assertEqual(delays, [5, 10, 15, 30])

def test_multiple_matches_fail_without_retry(self) -> None:
calls: list[tuple[str, str]] = []
delays: list[float] = []
fallback_calls: list[tuple[str, int]] = []

def lookup(repository: str, head: str) -> list[dict]:
calls.append((repository, head))
return [self.pr, {**self.pr, "number": 24}]

with self.assertRaisesRegex(
merge_queue_search.InputFault,
r"ambiguous_queue_member:.*:matches=2:attempts=1$",
):
merge_queue_search.member_pr(
"cortexkit/aft",
self.pr["head"]["sha"],
queue_ref_pr_number=23,
lookup=lookup,
queue_ref_lookup=lambda repository, number: fallback_calls.append((repository, number)),
sleep=delays.append,
)

self.assertEqual(len(calls), 1)
self.assertEqual(delays, [])
self.assertEqual(fallback_calls, [])

def test_lookup_http_error_is_not_retried(self) -> None:
calls = 0
delays: list[float] = []

def lookup(_repository: str, _head: str) -> list[dict]:
nonlocal calls
calls += 1
raise merge_queue_search.InputFault("queue_pr_lookup_failed:head:unauthorized")

with self.assertRaisesRegex(
merge_queue_search.InputFault, r"queue_pr_lookup_failed:head:unauthorized$"
):
merge_queue_search.member_pr(
"cortexkit/aft",
self.pr["head"]["sha"],
lookup=lookup,
sleep=delays.append,
)

self.assertEqual(calls, 1)
self.assertEqual(delays, [])

def test_queue_ref_pr_is_used_only_when_head_matches(self) -> None:
calls = 0

def lookup(_repository: str, _head: str) -> list[dict]:
nonlocal calls
calls += 1
return []

result = merge_queue_search.member_pr(
"cortexkit/aft",
self.pr["head"]["sha"],
queue_ref_pr_number=23,
lookup=lookup,
queue_ref_lookup=lambda _repository, _number: self.pr,
sleep=lambda _seconds: None,
)

self.assertEqual(result, self.pr)
self.assertEqual(calls, 5)

def test_queue_ref_pr_with_different_head_does_not_override_fault(self) -> None:
calls = 0

def lookup(_repository: str, _head: str) -> list[dict]:
nonlocal calls
calls += 1
return []

different_head = {**self.pr, "head": {**self.pr["head"], "sha": "b" * 40}}
with self.assertRaisesRegex(
merge_queue_search.InputFault,
r"ambiguous_queue_member:.*:matches=0:attempts=5$",
):
merge_queue_search.member_pr(
"cortexkit/aft",
self.pr["head"]["sha"],
queue_ref_pr_number=23,
lookup=lookup,
queue_ref_lookup=lambda _repository, _number: different_head,
sleep=lambda _seconds: None,
)

self.assertEqual(calls, 5)


if __name__ == "__main__":
unittest.main()
17 changes: 5 additions & 12 deletions crates/aft/src/access_policy_task_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -547,24 +547,17 @@ fn access_policy_foreign_worker_status_and_controls_are_denied() {
}

#[test]
fn access_policy_task_folder_cache_is_pinned_and_shared_between_files() {
fn access_policy_task_folder_cache_is_identity_checked_and_shared_between_files() {
let f = fixture();
crate::access_policy::with_scope(Some(f.head.clone()), || {
assert!(read_as(&f.ctx, "head-session", &f.worker.command).success);
std::fs::write(&f.worker.json, "invalid metadata").unwrap();
assert!(read_as(&f.ctx, "head-session", &f.worker.stdout).success);
// Windows refuses to rename a folder while the pinned record holds a
// handle to it, which already rules out the swap this test guards.
let renamed = std::fs::rename(&f.worker.dir, f.worker.dir.with_extension("old"));
if cfg!(windows) {
assert!(
renamed.is_err(),
"a pinned task folder must not be renamable"
);
return;
}
renamed.unwrap();
// Cached permissions retain identities, not handles. Renaming is allowed,
// but a new directory at the same path must not inherit those permissions.
std::fs::rename(&f.worker.dir, f.worker.dir.with_extension("old")).unwrap();
std::fs::create_dir_all(&f.worker.control_dir).unwrap();
std::fs::create_dir_all(&f.worker.io_dir).unwrap();
std::fs::write(&f.worker.command, "replacement").unwrap();
assert_floor(
read_as(&f.ctx, "head-session", &f.worker.command),
Expand Down
60 changes: 60 additions & 0 deletions crates/aft/src/bash_background/persistence.rs
Original file line number Diff line number Diff line change
Expand Up @@ -223,7 +223,66 @@ pub struct PinnedDir {
path: PathBuf,
}

/// A directory's identity without an open handle. Historical read permissions
/// reopen the directory and compare identities instead of pinning it forever.
#[derive(Debug, Clone)]
pub(crate) struct DirectorySnapshot {
path: PathBuf,
identity: DirectoryIdentity,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct DirectoryIdentity {
#[cfg(unix)]
device: u64,
#[cfg(unix)]
inode: u64,
#[cfg(windows)]
volume: u32,
#[cfg(windows)]
index: (u32, u32),
}

impl DirectorySnapshot {
pub(crate) fn reopen(&self) -> io::Result<PinnedDir> {
let current = PinnedDir::open(&self.path)?;
if current.identity()? != self.identity {
return Err(io::Error::new(
io::ErrorKind::PermissionDenied,
"registered directory path identity changed",
));
}
Ok(current)
}
}

impl PinnedDir {
fn identity(&self) -> io::Result<DirectoryIdentity> {
#[cfg(unix)]
{
let metadata = self.file.metadata()?;
Ok(DirectoryIdentity {
device: metadata.dev(),
inode: metadata.ino(),
})
}
#[cfg(windows)]
{
let info = windows_file_information(&self.file)?;
Ok(DirectoryIdentity {
volume: info.volume_serial_number,
index: (info.file_index_high, info.file_index_low),
})
}
}

pub(crate) fn snapshot(&self) -> io::Result<DirectorySnapshot> {
Ok(DirectorySnapshot {
path: self.path.clone(),
identity: self.identity()?,
})
}

pub fn open(path: &Path) -> io::Result<Self> {
#[cfg(test)]
work_counts::record_open();
Expand Down Expand Up @@ -467,6 +526,7 @@ impl PinnedDir {
}
}

#[cfg(windows)]
pub(crate) fn ensure_current_identity(&self) -> io::Result<()> {
let current = Self::open(&self.path)?;
if !self.same_identity(&current)? {
Expand Down
44 changes: 28 additions & 16 deletions crates/aft/src/bash_background/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -947,9 +947,10 @@ pub(crate) struct RegistryInner {
Mutex<Option<(std::sync::mpsc::Sender<()>, std::sync::mpsc::Receiver<()>)>>,
pub(crate) tasks: Mutex<HashMap<String, Arc<BgTask>>>,
/// Session owners of output files recovered from task records on disk.
/// Directory handles prevent cached permissions from following a replaced I/O directory.
/// Identity snapshots prevent cached permissions from following a replaced
/// I/O directory without keeping a descriptor open for each historical task.
persisted_artifact_owners:
Mutex<HashMap<PathBuf, (String, Arc<super::persistence::PinnedDir>)>>,
Mutex<HashMap<PathBuf, (String, super::persistence::DirectorySnapshot)>>,
task_read_records: Mutex<VecDeque<(PathBuf, Arc<task_access::TaskReadRecord>)>>,
#[cfg(test)]
task_access_work: task_access::WorkCounters,
Expand Down Expand Up @@ -1584,21 +1585,23 @@ impl BgTaskRegistry {
})
}

fn persisted_artifact_owner(
&self,
session_id: &str,
requested: &Path,
) -> Option<Arc<super::persistence::PinnedDir>> {
fn persisted_artifact_owner(&self, session_id: &str, requested: &Path) -> Option<()> {
let cached = self
.inner
.persisted_artifact_owners
.lock()
.ok()?
.get(requested)
.cloned();
if let Some((owner, io)) = cached {
return (owner == session_id && io.open_file(requested.file_name()?, false).is_ok())
.then_some(io);
if let Some((owner, snapshot)) = cached {
if owner != session_id {
return None;
}
let io = snapshot.reopen().ok()?;
return io
.open_file(requested.file_name()?, false)
.is_ok()
.then_some(());
}
// A task record may not have been loaded since the daemon restarted.
// Check the record named by the output path, not neighboring tasks or files.
Expand Down Expand Up @@ -1637,12 +1640,13 @@ impl BgTaskRegistry {
{
return None;
}
let io = Arc::clone(&resolved.dirs.io);
self.inner.persisted_artifact_owners.lock().ok()?.insert(
requested.to_path_buf(),
(session_id.to_string(), Arc::clone(&io)),
);
Some(io)
let snapshot = resolved.dirs.io.snapshot().ok()?;
self.inner
.persisted_artifact_owners
.lock()
.ok()?
.insert(requested.to_path_buf(), (session_id.to_string(), snapshot));
Some(())
}

pub fn read_artifact_path(
Expand Down Expand Up @@ -12521,6 +12525,14 @@ fn release_terminal_io_handles(task: &BgTask) -> bool {
_ => (None, false),
};
drop(handles);
// An adopted process's pidfd or kqueue is useful only until it becomes
// terminal. History must not keep the exit observer's descriptor alive.
#[cfg(unix)]
if runtime_released {
if let Ok(mut observer) = task.adopted_exit.lock() {
*observer = Default::default();
}
}
runtime_released && task.terminal_reader_done.load(Ordering::SeqCst)
}

Expand Down
Loading
Loading