okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petkernel: keep output once, bound snapshots and replays, drop the outbox scanf479db5

main
branchesmainall refs →
4.8 KiB117 linesraw
1"""Output retention invariants the daemon cannot observe through a cell result.
2
3A capture keeps the first RETAIN bytes and the last PREVIEW bytes of what was
4written, as bytes: a pipe chunk that splits a UTF-8 character must survive,
5the tail must equal the true end of the stream whichever path wrote it, and
6the memory held must stay within the two caps however much flows through.
7"""
8
9from pathlib import Path
10import sys
11import tempfile
12import tracemalloc
13from typing import cast
14import unittest
15
16sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "priv" / "python"))
17import albedo_state  # noqa: E402
18from albedo_capture import PREVIEW, RETAIN, Capture  # noqa: E402
19
20
21def stream_of(capture: Capture, pieces) -> bytes:
22    whole = bytearray()
23    for piece in pieces:
24        if isinstance(piece, str):
25            capture.write(piece)
26            whole += piece.encode("utf-8", errors="replace")
27        else:
28            capture.write_bytes(piece)
29            whole += piece
30    return bytes(whole)
31
32
33class CaptureTest(unittest.TestCase):
34    def test_a_character_split_across_pipe_chunks_is_kept_whole(self):
35        capture = Capture("job", "job")
36        text = "héllo wörld — ✓\n".encode()
37        for cut in range(1, len(text)):
38            capture.write_bytes(text[:cut])
39            capture.write_bytes(text[cut:])
40        self.assertEqual(capture.read(0, RETAIN), text.decode() * (len(text) - 1))
41        self.assertNotIn("\ufffd", capture.tail().decode())
42
43    def test_both_write_paths_agree_on_what_is_kept(self):
44        for pieces in (
45            ["x" * (RETAIN + 3 * PREVIEW)],  # one ASCII print past every cap
46            ["x" * 10, "y" * (RETAIN + PREVIEW + 1), "tail"],
47            ["ü" * (RETAIN + 10)],  # two bytes per character: the full encode
48            [b"\xff" * 100, "ascii" * 100, b"\xe2\x9c\x93" * (RETAIN // 3)],
49            ["short", b"bytes"],
50        ):
51            with self.subTest(pieces=[len(p) for p in pieces]):
52                capture = Capture("c")
53                whole = stream_of(capture, pieces)
54                self.assertEqual(capture.seen, len(whole))
55                self.assertEqual(bytes(capture.data), whole[:RETAIN])
56                self.assertEqual(capture.tail(), whole[-PREVIEW:])
57                self.assertEqual(capture.tail(16), whole[-16:])
58                self.assertEqual(capture.tail(0), b"")
59
60    def test_retained_bytes_stay_within_the_two_caps(self):
61        for label, pieces in (
62            ("pipe chunks", [bytes(range(256)) * 1000] * 40),
63            ("small prints", ["some line of output\n"] * 300_000),
64            ("one print", ["x" * (3 * RETAIN)]),
65        ):
66            with self.subTest(label):
67                capture = Capture("c")
68                whole = stream_of(capture, pieces)
69                self.assertEqual(capture.seen, len(whole))
70                self.assertEqual(len(capture.data), RETAIN)
71                # bytearray growth slack is bounded; the buffers themselves are capped.
72                held = sys.getsizeof(capture.data) + sys.getsizeof(capture._tail)
73                self.assertLess(held, RETAIN + PREVIEW + PREVIEW // 4)
74
75    def test_a_tail_buffer_exists_only_once_the_start_is_full(self):
76        capture = Capture("c")
77        capture.write("a" * (RETAIN - 1))
78        self.assertIsNone(capture._tail)
79        self.assertEqual(capture.tail(4), b"aaaa")
80        capture.write("bb")
81        self.assertIsNotNone(capture._tail)
82        self.assertEqual(capture.tail(4), b"aabb")
83        self.assertEqual(len(capture.data), RETAIN)
84
85    def test_a_huge_ascii_print_is_not_encoded_whole(self):
86        capture = Capture("c")
87        text = "z" * (64 * 1024 * 1024)
88        tracemalloc.start()
89        try:
90            capture.write(text)
91            _, peak = tracemalloc.get_traced_memory()
92        finally:
93            tracemalloc.stop()
94        self.assertEqual(capture.seen, len(text))
95        # The transient is a few retained-size pieces, not the 64 MiB text encoded.
96        self.assertLess(peak, 4 * RETAIN)
97
98
99class SnapshotBoundTest(unittest.TestCase):
100    def test_an_oversized_value_costs_the_cap_not_its_size(self):
101        namespace: dict[str, object] = {"big": list(range(4_000_000)), "small": [1]}
102        target = Path(self.enterContext(tempfile.TemporaryDirectory()))
103        tracemalloc.start()
104        try:
105            state = albedo_state.save_state(str(target / "state"), namespace, {}, set())
106            _, peak = tracemalloc.get_traced_memory()
107        finally:
108            tracemalloc.stop()
109        skipped = cast(list[dict[str, str]], state["skipped"])
110        self.assertEqual(state["saved"], ["small"])
111        self.assertEqual([item["name"] for item in skipped], ["big"])
112        self.assertIn("per-variable cap", skipped[0]["reason"])
113        self.assertLess(peak, 3 * albedo_state.STATE_MAX_VALUE)
114
115
116if __name__ == "__main__":
117    unittest.main()