rho-coding-agent 2.11.0

A fast Rust agent harness with a small footprint and opinionated defaults
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use std::{cell::RefCell, sync::mpsc};

use pretty_assertions::assert_eq;
use rho_providers::model::Message;
use tempfile::TempDir;

use super::*;
use crate::session::Session;

struct MigrationWait {
    blocked: mpsc::Sender<()>,
    resume: mpsc::Receiver<()>,
}

thread_local! {
    static MIGRATION_WAIT: RefCell<Option<MigrationWait>> = const { RefCell::new(None) };
}

// Covers: a waiting opener must not invalidate evidence rebuilt by the first
// migration. Owner: search index SQLite migration, with actual lock contention.
#[test]
fn concurrent_migration_preserves_rebuilt_cache() {
    let root = TempDir::new().unwrap();
    let cwd = TempDir::new().unwrap();
    let session = Session::create_in_root(root.path(), cwd.path()).unwrap();
    session
        .append_message(&Message::user_text("migrationneedle"))
        .unwrap();
    let cancellation = CancellationToken::new();
    let mut first = open(root.path()).unwrap();
    refresh(&mut first, root.path(), false, &cancellation).unwrap();
    first.pragma_update(None, "user_version", 1).unwrap();
    let (blocked_tx, blocked_rx) = mpsc::channel();
    let (resume_tx, resume_rx) = mpsc::channel();
    std::thread::scope(|scope| {
        let lock = first
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
            .unwrap();
        let root = root.path();
        let second = scope.spawn(move || {
            MIGRATION_WAIT.with(|wait| {
                *wait.borrow_mut() = Some(MigrationWait {
                    blocked: blocked_tx,
                    resume: resume_rx,
                });
            });
            let mut connection = Connection::open(root.join("search.sqlite3")).unwrap();
            connection
                .busy_handler(Some(|_| {
                    MIGRATION_WAIT.with(|wait| {
                        let Some(wait) = wait.borrow_mut().take() else {
                            return false;
                        };
                        // Use the store's existing lock budget only as a failure
                        // bound. Channels, not elapsed time, order the migration.
                        wait.blocked.send(()).is_ok()
                            && wait
                                .resume
                                .recv_timeout(crate::sqlite_support::BUSY_TIMEOUT)
                                .is_ok()
                    })
                }))
                .unwrap();
            migrate(&mut connection).unwrap();
            refresh(&mut connection, root, false, &CancellationToken::new()).unwrap()
        });
        blocked_rx
            .recv_timeout(crate::sqlite_support::BUSY_TIMEOUT)
            .unwrap();
        lock.rollback().unwrap();
        migrate(&mut first).unwrap();
        let rebuilt = refresh(&mut first, root, false, &cancellation).unwrap();
        assert_eq!(rebuilt.files_updated, 1);
        resume_tx.send(()).unwrap();
        let warm = second.join().unwrap();
        assert_eq!(
            (warm.reconciled, warm.files_checked, warm.bytes_read),
            (false, 0, 0)
        );
        assert_eq!(
            first
                .query_row(
                    "select count(*) from evidence_fts where evidence_fts match 'migrationneedle'",
                    [],
                    |row| row.get::<_, usize>(0)
                )
                .unwrap(),
            1
        );
    });
}