asc-engine 0.2.0

//! The engine's use of the disk cache tiers beyond inflated DEX entries, and
//! the memory budget shared by the resident hosts of one user.
//!
//! * L2 (`asc_cache::Tier::Derived`): the locator tables of each logical DEX
//!   ([`LOCATORS_KIND`]). A search first adopts the tables a previous
//!   process persisted (mapped, so every process shares one copy of their
//!   pages), and queues the tables it had to build for writing once they took
//!   long enough to build (`DiskCache::admits_derived`). Once written, the copy in the memory cache is
//!   swapped for a view of the file, like inflated bytes are remapped to L1.
//! * L3 (`asc_cache::Tier::Rendered`): the result rows of a `find_refs`
//!   query, replayed by a later run of the same query on the same APK with
//!   the same build of the tool.
//!
//! * L2 also keeps the class graph (`hierarchy::GRAPH_KIND`, one per APK),
//!   used in place where it is mapped.
//!
//! Other derived indexes (call graph edges, resource tables, manifest
//! summaries) go to L2 the same way: serialise them into bytes that can be
//! viewed in place, store them with `DiskCache::put_derived(key, name, kind,
//! version, build_cost, bytes)` from the cache writer, and map them back with
//! `DiskCache::get_derived(key, name, kind, version)`. Pick a `kind` of their
//! own and bump `version` whenever the layout changes; per-APK indexes use a
//! fixed `name` such as `apk`.

use crate::{
    CachedEntry, DexBytes, FindOptions, LogicalDex, Matched, MemoryPolicy, RefBatch, RefResult, RefSite, Shape, SharedMemory, TextRows,
    available_memory,
};
use asc_cache::{ApkKey, DiskCache, LockExt, ProcessRegistry};
use asc_dex::Dex;
use asc_findrefs::{ClassRelation, Query, SearchState};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

/// L2 kind of the persisted locator tables.
pub(crate) const LOCATORS_KIND: &str = "locators";
/// `find_refs` results larger than this (encoded) are not kept in L3.
pub(crate) const L3_MAX_RESULT: usize = 4 << 20;
/// Text rows ([`Shape::Text`]) larger than this are not kept in L3, nor
/// beyond an eighth of the disk cache: a few broad queries must not evict the
/// inflated DEX files every query needs.
const L3_MAX_TEXT: usize = 32 << 20;
/// Layout version of the encoded `find_refs` results.
const RESULT_FORMAT: u32 = 6;
/// Ends the encoded batches (where an entry name's length would be), so a
/// truncated file is told from a complete one; the run's notes follow it.
const RESULT_END: u32 = u32::MAX;

/// L2 bookkeeping of one logical DEX, shared by the entries that replace
/// each other in the memory cache.
#[derive(Default)]
pub(crate) struct Persisted {
    /// L2 has been consulted for this DEX.
    tried: AtomicBool,
    /// Tables on disk or queued for writing (`SearchState::table_mask` bits).
    queued: AtomicU32,
    /// Time spent building the tables so far.
    cost_us: AtomicU64,
}

/// Adopt the locator tables persisted for `l` (once per logical DEX).
pub(crate) fn load_derived(disk: &DiskCache, key: &ApkKey, l: &LogicalDex, dex: &Dex<'_>) -> bool {
    if l.persisted.tried.swap(true, Ordering::Relaxed) {
        return false;
    }
    let Some(d) = disk.get_derived(key, &l.name, LOCATORS_KIND, asc_findrefs::PERSIST_FORMAT_VERSION) else { return false };
    let cost = d.build_cost.as_micros() as u64;
    let Some(state) = SearchState::from_persisted(dex, Arc::new(d)) else { return false };
    l.persisted.queued.fetch_or(state.table_mask(), Ordering::Relaxed);
    l.persisted.cost_us.fetch_add(cost, Ordering::Relaxed);
    l.state.fill_from(state);
    true
}

/// Tables of a logical DEX to persist.
pub(crate) struct DerivedJob {
    /// Memory cache entry and logical index the tables belong to.
    entry: String,
    logical: usize,
    name: String,
    data: DexBytes,
    state: Arc<SearchState>,
    persisted: Arc<Persisted>,
    /// The `queued` bits this job claimed.
    claimed: u32,
}

impl DerivedJob {
    /// Heap the job keeps alive until written: the tables it persists.
    pub(crate) fn held_bytes(&self) -> usize {
        self.state.heap_bytes()
    }

    /// Not written after all: a later search may queue these tables again.
    pub(crate) fn abandon(self) {
        self.persisted.queued.fetch_and(!self.claimed, Ordering::Relaxed);
    }
}

/// A write of the tables of `entry.logical()[idx]`, when a search that
/// started with `mask_before` built tables worth persisting (`spent` is
/// what it took).
pub(crate) fn derived_job(disk: &DiskCache, entry: &CachedEntry, idx: usize, mask_before: u32, spent: Duration) -> Option<DerivedJob> {
    let l = entry.logical().get(idx)?;
    let mask = l.state.table_mask();
    let p = &l.persisted;
    if mask == mask_before && mask & !p.queued.load(Ordering::Relaxed) == 0 {
        return None;
    }
    let cost = if mask != mask_before {
        p.cost_us.fetch_add(spent.as_micros() as u64, Ordering::Relaxed) + spent.as_micros() as u64
    } else {
        p.cost_us.load(Ordering::Relaxed)
    };
    if !disk.admits_derived(Duration::from_micros(cost)) {
        return None;
    }
    let claimed = mask & !p.queued.fetch_or(mask, Ordering::Relaxed);
    if claimed == 0 {
        return None;
    }
    Some(DerivedJob {
        entry: entry.name.clone(),
        logical: idx,
        name: l.name.clone(),
        data: Arc::clone(&l.data),
        state: Arc::clone(&l.state),
        persisted: Arc::clone(&l.persisted),
        claimed,
    })
}

/// Write a [`DerivedJob`] (on the cache writer thread), then swap the heap
/// tables of the memory cache's entry for views of the file.
pub(crate) fn write_derived(disk: &DiskCache, memory: Option<&SharedMemory>, key: &ApkKey, job: DerivedJob) {
    let Ok(dex) = Dex::with_name(&job.data, &job.name) else { return };
    let mask = job.state.table_mask();
    let bytes = job.state.persist(&dex);
    let cost = Duration::from_micros(job.persisted.cost_us.load(Ordering::Relaxed));
    let version = asc_findrefs::PERSIST_FORMAT_VERSION;
    if disk.put_derived(key, &job.name, LOCATORS_KIND, version, cost, &bytes).is_err() {
        return;
    }
    drop(bytes);
    let Some(memory) = memory else { return };
    let Some(mapped) =
        disk.get_derived(key, &job.name, LOCATORS_KIND, version).and_then(|d| SearchState::from_persisted(&dex, Arc::new(d)))
    else {
        return;
    };
    // Tables built after the snapshot would be lost by the swap: keep the
    // heap state then (the next write maps them all).
    if mapped.table_mask() != mask || job.state.table_mask() != mask {
        return;
    }
    let mut mem = crate::lock_memory(memory);
    let Some(current) = mem.peek(key, &job.entry) else { return };
    let still_ours = current.logical.get().and_then(|l| l.get(job.logical)).is_some_and(|l| Arc::ptr_eq(&l.state, &job.state));
    if still_ours {
        let new = Arc::new(current.with_state(job.logical, Arc::new(mapped)));
        if mem.replace(key, &job.entry, &current, new) {
            mem.recharge(key, &job.entry, crate::CachedEntry::charged_bytes);
        }
    }
}

/// Fingerprint of this build of the tool: the crate version plus the size
/// and modification time of the running executable, so a rebuilt binary
/// never replays results of the old one.
fn tool_version() -> &'static str {
    static VERSION: OnceLock<String> = OnceLock::new();
    VERSION.get_or_init(|| {
        let exe = std::env::current_exe().ok().and_then(|p| std::fs::metadata(p).ok());
        let stamp = exe.map(|m| {
            let mtime = m.modified().ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map_or(0, |d| d.as_nanos());
            format!("{}.{mtime}", m.len())
        });
        format!("{}+{}", env!("CARGO_PKG_VERSION"), stamp.unwrap_or_default())
    })
}

/// The L3 key of a `find_refs` query.
pub(crate) fn findrefs_key(query: &Query, opts: FindOptions, shape: Shape) -> String {
    let query = match shape {
        Shape::Results => format!("{query:?}\0{opts:?}"),
        Shape::Text => format!("{query:?}\0{opts:?}\0text"),
    };
    asc_cache::rendered_key("findrefs", tool_version(), &query)
}

/// The most bytes a `find_refs` result of `shape` may take in `disk`'s L3.
pub(crate) fn result_limit(disk: &DiskCache, shape: Shape) -> usize {
    match shape {
        Shape::Results => L3_MAX_RESULT,
        Shape::Text => usize::try_from(disk.capacity_for(0) / 8).unwrap_or(usize::MAX).min(L3_MAX_TEXT),
    }
}

/// Encodes the batches of a `find_refs` run for L3 as they stream past.
/// Runs with an error (an entry that could not be searched may succeed next
/// time) or results beyond the recorder's limit ([`result_limit`]) are not
/// kept.
pub(crate) struct ResultRecorder {
    out: Vec<u8>,
    keep: bool,
    limit: usize,
}

impl ResultRecorder {
    pub(crate) fn new(limit: usize) -> ResultRecorder {
        let mut out = Vec::new();
        out.extend_from_slice(&RESULT_FORMAT.to_le_bytes());
        ResultRecorder { out, keep: true, limit }
    }

    /// Stop recording: the run will not be kept.
    fn abandon(&mut self) {
        self.keep = false;
        self.out = Vec::new();
    }

    pub(crate) fn record(&mut self, b: &RefBatch) {
        if !self.keep {
            return;
        }
        // Text rows are the bulk of a broad run: check them before copying.
        let text_bytes = b.text.as_ref().map_or(0, |t| t.as_str().len() + 4 * t.len());
        if b.error.is_some() || self.out.len().saturating_add(text_bytes) > self.limit {
            self.abandon();
            return;
        }
        put_str(&mut self.out, &b.entry_name);
        put_u32(&mut self.out, b.results.len() as u32);
        for r in &b.results {
            put_str(&mut self.out, &r.dex_name);
            put_u32(&mut self.out, r.method_idx);
            put_str(&mut self.out, &r.method);
            put_str(&mut self.out, &r.signature);
            put_u32(&mut self.out, r.matched.len() as u32);
            for m in &r.matched {
                self.out.push(match m {
                    Matched::String(_) => 0,
                    Matched::Symbol(_) => 1,
                    Matched::Field { .. } => 2,
                });
                put_str(&mut self.out, m.as_str());
                if let Matched::Field { access, .. } = m {
                    self.out.push(match access {
                        asc_findrefs::FieldAccess::Read => 1,
                        asc_findrefs::FieldAccess::Write => 2,
                    });
                }
            }
            self.out.push(match r.relation {
                None => 0,
                Some(ClassRelation::Extends) => 1,
                Some(ClassRelation::Implements) => 2,
            });
            put_u32(&mut self.out, r.sites.len() as u32);
            for site in &r.sites {
                put_u32(&mut self.out, site.offset);
                put_u32(&mut self.out, site.matched as u32);
                self.out.push(match site.field_access {
                    None => 0,
                    Some(asc_findrefs::FieldAccess::Read) => 1,
                    Some(asc_findrefs::FieldAccess::Write) => 2,
                });
                put_opt(&mut self.out, site.pc);
                put_opt(&mut self.out, site.line);
            }
            match &r.context {
                None => self.out.push(0),
                Some(c) => {
                    self.out.push(1);
                    put_str(&mut self.out, c);
                }
            }
        }
        match &b.text {
            None => self.out.push(0),
            Some(t) => {
                let Ok(ends) = t.ends().iter().map(|&e| u32::try_from(e)).collect::<Result<Vec<_>, _>>() else {
                    return self.abandon();
                };
                self.out.push(1);
                put_str(&mut self.out, t.as_str());
                put_u32(&mut self.out, ends.len() as u32);
                for e in ends {
                    put_u32(&mut self.out, e);
                }
            }
        }
        if self.out.len() > self.limit {
            self.abandon();
        }
    }

    /// The encoded run, its `notes` ([`crate::RefSummary::notes`]) last.
    pub(crate) fn finish(mut self, notes: &[String]) -> Option<Vec<u8>> {
        put_u32(&mut self.out, RESULT_END);
        put_u32(&mut self.out, notes.len() as u32);
        for n in notes {
            put_str(&mut self.out, n);
        }
        self.keep.then_some(self.out)
    }
}

fn put_u32(out: &mut Vec<u8>, v: u32) {
    out.extend_from_slice(&v.to_le_bytes());
}

fn put_opt(out: &mut Vec<u8>, v: Option<u32>) {
    match v {
        None => out.push(0),
        Some(v) => {
            out.push(1);
            put_u32(out, v);
        }
    }
}

fn put_str(out: &mut Vec<u8>, s: &str) {
    put_u32(out, s.len() as u32);
    out.extend_from_slice(s.as_bytes());
}

/// A run [`ResultRecorder`] kept.
pub(crate) struct Recorded {
    pub batches: Vec<RefBatch>,
    pub notes: Vec<String>,
}

/// The run recorded by [`ResultRecorder`]; `None` for anything malformed.
/// Text rows stay views of `payload` (typically a mapped L3 file).
pub(crate) fn decode_batches(payload: impl AsRef<[u8]> + Send + Sync + 'static) -> Option<Recorded> {
    let payload: Arc<dyn AsRef<[u8]> + Send + Sync> = Arc::new(payload);
    let bytes = (*payload).as_ref();
    let mut r = Cursor { bytes, at: 0 };
    if r.u32()? != RESULT_FORMAT {
        return None;
    }
    let mut out = Vec::new();
    loop {
        let len = r.u32()?;
        if len == RESULT_END {
            let notes = (0..r.u32()?).map(|_| r.str()).collect::<Option<Vec<_>>>()?;
            return (r.at == bytes.len()).then_some(Recorded { batches: out, notes });
        }
        let entry_name = String::from_utf8(r.take(len as usize)?.to_vec()).ok()?;
        let n = r.u32()? as usize;
        let mut results = Vec::with_capacity(n.min(bytes.len()));
        for _ in 0..n {
            let dex_name = r.str()?;
            let method_idx = r.u32()?;
            let method = r.str()?;
            let signature = r.str()?;
            let m = r.u32()? as usize;
            let matched = (0..m)
                .map(|_| match r.u8()? {
                    0 => Some(Matched::String(r.str()?)),
                    1 => Some(Matched::Symbol(r.str()?)),
                    2 => {
                        let symbol = r.str()?;
                        let access = match r.u8()? {
                            1 => asc_findrefs::FieldAccess::Read,
                            2 => asc_findrefs::FieldAccess::Write,
                            _ => return None,
                        };
                        Some(Matched::Field { symbol, access })
                    }
                    _ => None,
                })
                .collect::<Option<Vec<_>>>()?;
            let relation = match r.u8()? {
                0 => None,
                1 => Some(ClassRelation::Extends),
                2 => Some(ClassRelation::Implements),
                _ => return None,
            };
            let sites = (0..r.u32()?)
                .map(|_| {
                    let offset = r.u32()?;
                    let matched = r.u32()? as usize;
                    let field_access = match r.u8()? {
                        0 => None,
                        1 => Some(asc_findrefs::FieldAccess::Read),
                        2 => Some(asc_findrefs::FieldAccess::Write),
                        _ => return None,
                    };
                    Some(RefSite { offset, matched, field_access, pc: r.opt()?, line: r.opt()? })
                })
                .collect::<Option<Vec<_>>>()?;
            let context = match r.u8()? {
                0 => None,
                1 => Some(r.str()?),
                _ => return None,
            };
            results.push(RefResult { dex_name, method_idx, method, signature, matched, relation, sites, context });
        }
        let text = match r.u8()? {
            0 => None,
            1 => {
                let len = r.u32()? as usize;
                let start = r.at;
                r.take(len)?;
                let ends = (0..r.u32()?).map(|_| r.u32().map(|e| e as usize)).collect::<Option<Vec<_>>>()?;
                Some(TextRows::shared(Arc::clone(&payload), start..start + len, ends)?)
            }
            _ => return None,
        };
        out.push(RefBatch { entry_name, results, text, inflate_us: 0, search_us: 0, source: "result", error: None });
    }
}

struct Cursor<'b> {
    bytes: &'b [u8],
    at: usize,
}

impl Cursor<'_> {
    fn take(&mut self, n: usize) -> Option<&[u8]> {
        let s = self.bytes.get(self.at..self.at.checked_add(n)?)?;
        self.at += n;
        Some(s)
    }

    fn u8(&mut self) -> Option<u8> {
        Some(self.take(1)?[0])
    }

    fn u32(&mut self) -> Option<u32> {
        Some(u32::from_le_bytes(self.take(4)?.try_into().ok()?))
    }

    /// `Some(None)` / `Some(Some(v))` for a present field, `None` when malformed.
    fn opt(&mut self) -> Option<Option<u32>> {
        match self.u8()? {
            0 => Some(None),
            1 => Some(Some(self.u32()?)),
            _ => None,
        }
    }

    fn str(&mut self) -> Option<String> {
        let n = self.u32()? as usize;
        String::from_utf8(self.take(n)?.to_vec()).ok()
    }
}

/// How often a memory cache re-reads the registry and resizes itself.
const REBALANCE_EVERY: Duration = Duration::from_secs(1);

/// The process's registration: one per process, however many caches it has.
struct Registration {
    caches: usize,
    registry: Option<ProcessRegistry>,
    /// Bytes held by all of this process's memory caches.
    held: u64,
    /// Requests in progress ([`Busy`]).
    requests: usize,
}

impl Registration {
    fn publish(&self) {
        if let Some(registry) = &self.registry {
            let _ = registry.update(self.held, self.requests > 0);
        }
    }
}

static REGISTRATION: Mutex<Registration> = Mutex::new(Registration { caches: 0, registry: None, held: 0, requests: 0 });

/// Marks this process busy in the user's process registry while a request
/// (an MCP tool call, a daemon command) runs, so the others split the CPUs
/// with it from its first moment ([`auto_threads`](crate::auto_threads)) and take them back as soon
/// as it ends. Nested and concurrent requests count once.
#[must_use = "the process is busy only while the guard lives"]
pub struct Busy(());

impl Busy {
    /// Mark the process busy until the guard is dropped.
    ///
    /// # Panics
    ///
    /// When the process registration mutex is poisoned (another thread
    /// panicked while holding it).
    pub fn enter() -> Busy {
        let mut reg = REGISTRATION.lock().expect("process registration mutex poisoned");
        reg.requests += 1;
        if reg.requests == 1 {
            reg.publish();
        }
        Busy(())
    }
}

impl Drop for Busy {
    fn drop(&mut self) {
        let mut reg = REGISTRATION.lock().expect("process registration mutex poisoned");
        reg.requests -= 1;
        if reg.requests == 0 {
            reg.publish();
        }
    }
}

/// The memory budget of one memory cache: published to the user's process
/// registry and, when automatic, resized to this process's share of one
/// budget for all registered processes ([`MemoryPolicy::share`]).
pub(crate) struct MemoryBudget {
    auto: Option<MemoryPolicy>,
    last: Mutex<Option<Instant>>,
    published: AtomicU64,
}

impl MemoryBudget {
    pub(crate) fn new(auto: Option<MemoryPolicy>) -> MemoryBudget {
        let mut reg = REGISTRATION.lock().expect("process registration mutex poisoned");
        reg.caches += 1;
        if reg.registry.is_none() {
            reg.registry = ProcessRegistry::default_dir().and_then(ProcessRegistry::open);
        }
        MemoryBudget { auto, last: Mutex::new(None), published: AtomicU64::new(0) }
    }

    /// Publish what `memory` holds and, when automatic, resize it to this
    /// process's share. At most once per [`REBALANCE_EVERY`].
    pub(crate) fn maybe_rebalance(&self, memory: &SharedMemory) {
        {
            let mut last = self.last.lock_or_default();
            if last.is_some_and(|t| t.elapsed() < REBALANCE_EVERY) {
                return;
            }
            *last = Some(Instant::now());
        }
        let used = {
            let m = crate::lock_memory(memory);
            (m.used_bytes() + m.reserved_bytes()) as u64
        };
        let previous = self.published.swap(used, Ordering::Relaxed);
        let (held, others) = {
            let mut reg = REGISTRATION.lock().expect("process registration mutex poisoned");
            reg.held = (reg.held + used).saturating_sub(previous);
            reg.publish();
            let Some(registry) = &reg.registry else { return };
            (reg.held, registry.others())
        };
        if let Some(policy) = self.auto {
            let cap = policy.share(available_memory(), held, &others);
            crate::lock_memory(memory).set_capacity(cap);
        }
    }
}

impl Drop for MemoryBudget {
    fn drop(&mut self) {
        let mut reg = REGISTRATION.lock().expect("process registration mutex poisoned");
        reg.held = reg.held.saturating_sub(self.published.load(Ordering::Relaxed));
        reg.caches -= 1;
        if reg.caches == 0 {
            // Unregisters (removes this process's entry).
            reg.registry = None;
        } else {
            reg.publish();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn results_roundtrip_and_errors_are_not_kept() {
        let row = |rel| RefResult {
            dex_name: "classes.dex".into(),
            method_idx: 7,
            method: "La;->b".into(),
            signature: "La;->b()V".into(),
            matched: vec![
                Matched::String("x\ny".into()),
                Matched::Symbol("La;->c:I".into()),
                Matched::Field { symbol: "La;->f:I".into(), access: asc_findrefs::FieldAccess::Read },
            ],
            relation: rel,
            sites: vec![
                RefSite { offset: 9, matched: 1, field_access: None, pc: Some(3), line: None },
                RefSite { offset: 11, matched: 2, field_access: Some(asc_findrefs::FieldAccess::Read), pc: Some(4), line: Some(8) },
            ],
            context: rel.map(|_| "    const-string v0, \"x\"".to_string()),
        };
        let batch = |name: &str, results, error: Option<String>| RefBatch {
            entry_name: name.into(),
            results,
            text: None,
            inflate_us: 5,
            search_us: 6,
            source: "inflate",
            error,
        };
        let mut rec = ResultRecorder::new(L3_MAX_RESULT);
        rec.record(&batch("classes.dex", vec![row(None), row(Some(ClassRelation::Implements))], None));
        rec.record(&batch("classes2.dex", vec![], None));
        let bytes = rec.finish(&["a note".to_string()]).unwrap();
        let Recorded { batches: back, notes } = decode_batches(bytes.clone()).unwrap();
        assert_eq!(notes, ["a note"]);
        assert_eq!(back.len(), 2);
        assert_eq!(back[0].results, vec![row(None), row(Some(ClassRelation::Implements))]);
        assert_eq!((back[1].entry_name.as_str(), back[1].source), ("classes2.dex", "result"));
        assert!(decode_batches(bytes[..bytes.len() - 1].to_vec()).is_none());
        // Cut at a batch boundary: the end marker is missing.
        let mut one = ResultRecorder::new(L3_MAX_RESULT);
        one.record(&batch("classes2.dex", vec![], None));
        let one = one.finish(&[]).unwrap();
        assert!(decode_batches(one[..one.len() - 4].to_vec()).is_none());
        assert_eq!(decode_batches(one).unwrap().batches.len(), 1);
        assert!(decode_batches(Vec::new()).is_none());
        let mut failed = ResultRecorder::new(L3_MAX_RESULT);
        failed.record(&batch("classes.dex", vec![], Some("bad".into())));
        assert!(failed.finish(&[]).is_none());
    }

    #[test]
    fn text_rows_roundtrip_within_their_limit() {
        let text = |lines: &[&str]| {
            let bytes: Arc<dyn AsRef<[u8]> + Send + Sync> =
                Arc::new(lines.iter().map(|l| format!("{l}\n")).collect::<String>().into_bytes());
            let mut ends = Vec::new();
            let mut at = 0;
            for l in lines {
                at += l.len();
                ends.push(at);
                at += 1;
            }
            let len = (*bytes).as_ref().len();
            TextRows::shared(bytes, 0..len, ends).unwrap()
        };
        let batch = |name: &str, t: TextRows| RefBatch {
            entry_name: name.into(),
            results: Vec::new(),
            text: Some(t),
            inflate_us: 0,
            search_us: 0,
            source: "inflate",
            error: None,
        };
        let a = text(&["classes.dex | La;->b()V | \"x\"", "classes.dex | La;->c()V | \"y\""]);
        let mut rec = ResultRecorder::new(1 << 10);
        rec.record(&batch("classes.dex", a.clone()));
        rec.record(&batch("classes2.dex", text(&[])));
        let Recorded { batches: back, notes } = decode_batches(rec.finish(&["n".to_string()]).unwrap()).unwrap();
        assert_eq!(notes, ["n"]);
        assert_eq!(back[0].text.as_ref(), Some(&a));
        assert_eq!(back[0].row_count(), 2);
        assert_eq!(back[1].text.as_ref().map(TextRows::len), Some(0));
        // Past the limit nothing is kept, and the rows are not copied first.
        let mut rec = ResultRecorder::new(64);
        rec.record(&batch("classes.dex", a));
        assert!(rec.out.is_empty() && rec.finish(&[]).is_none());
    }
}