1package vt
2
3import (
4 "bytes"
5 "compress/zlib"
6 "errors"
7 "fmt"
8 "io"
9 "sync"
10
11 renderer "github.com/bnema/vev-vt/core"
12)
13
14// ErrHistoryCorrupt reports corruption of an internally compressed sealed page.
15// Public history decoders reject malformed external data before installing it.
16var ErrHistoryCorrupt = errors.New("corrupt compressed history page")
17
18const maxHistoryPageEncodedBytes = 256 << 20
19
20// sealedPage owns immutable semantic contents and mutable physical backing.
21// Evicted-prefix wrappers and borrowed views share this owner. A returned Frame
22// is an internal read-only borrow: dropping the cached pointer cannot invalidate
23// an in-flight reader because Go keeps that reader's page reference alive.
24type sealedPage struct {
25 mu sync.Mutex
26 frame renderer.Frame
27 compressed []byte
28 encodedSize int
29 width, height int
30 reads, observed uint64
31 observedOnce bool
32 incompressible bool
33 restores uint64
34}
35
36func newSealedPage(frame renderer.Frame) *sealedPage {
37 return &sealedPage{frame: frame, width: frame.Width, height: frame.Height}
38}
39
40func (p *sealedPage) readFrame(cache bool) (renderer.Frame, error) {
41 p.mu.Lock()
42 defer p.mu.Unlock()
43 p.reads++
44 if p.frame.Height != 0 {
45 return p.frame, nil
46 }
47 frame, err := p.decode()
48 if err != nil {
49 return renderer.Frame{}, err
50 }
51 p.restores++
52 if cache {
53 p.frame = frame
54 }
55 return frame, nil
56}
57
58// decode is called with the page lock held. The exact uncompressed length and
59// zlib checksum are validated before VTC1's full semantic preflight runs.
60func (p *sealedPage) decode() (renderer.Frame, error) {
61 invalid := func(err error) (renderer.Frame, error) {
62 return renderer.Frame{}, fmt.Errorf("%w: %v", ErrHistoryCorrupt, err)
63 }
64 if p.encodedSize < historyHeaderBytes || p.encodedSize > maxHistoryPageEncodedBytes || len(p.compressed) == 0 {
65 return invalid(errors.New("missing backing"))
66 }
67 source := bytes.NewReader(p.compressed)
68 r, err := pageDecompressor(source)
69 if err != nil {
70 return invalid(err)
71 }
72 defer pageDecompressorPool.Put(r)
73 // Read exactly the recorded size, then require a clean end of stream (which
74 // also verifies the zlib checksum) with no trailing compressed bytes.
75 data := make([]byte, p.encodedSize)
76 if _, err := io.ReadFull(r, data); err != nil {
77 return invalid(err)
78 }
79 var extra [1]byte
80 if n, err := r.Read(extra[:]); n != 0 || err != io.EOF {
81 if err == nil || err == io.EOF {
82 err = errors.New("backing length mismatch")
83 }
84 return invalid(err)
85 }
86 if err := r.Close(); err != nil {
87 return invalid(err)
88 }
89 if source.Len() != 0 {
90 return invalid(errors.New("backing length mismatch"))
91 }
92 view, err := UnmarshalHistory(data)
93 if err != nil {
94 return invalid(err)
95 }
96 if len(view.chunks) != 1 {
97 return invalid(errors.New("unexpected page count"))
98 }
99 frame := view.chunks[0].page.frame
100 if frame.Width != p.width || frame.Height != p.height {
101 return invalid(errors.New("page geometry mismatch"))
102 }
103 return frame, nil
104}
105
106func (c *HistoryChunk) frameView() renderer.Frame {
107 frame, err := c.page.readFrame(true)
108 if err != nil {
109 panic(err)
110 } // Never substitute blank data for corrupted history.
111 return frame
112}
113
114// Restore validates and caches the chunk's backing before a caller starts a
115// read transaction. Ordinary reads restore transparently; if private backing is
116// corrupted they panic with ErrHistoryCorrupt rather than silently losing text.
117func (c *HistoryChunk) Restore() error {
118 if c == nil {
119 return nil
120 }
121 _, err := c.page.readFrame(true)
122 return err
123}
124
125// pageDecompressorPool reuses zlib readers; each one owns a 32 KiB window.
126var pageDecompressorPool sync.Pool
127
128func pageDecompressor(source io.Reader) (io.ReadCloser, error) {
129 if pooled, ok := pageDecompressorPool.Get().(io.ReadCloser); ok {
130 if err := pooled.(zlib.Resetter).Reset(source, nil); err != nil {
131 return nil, err
132 }
133 return pooled, nil
134 }
135 return zlib.NewReader(source)
136}
137
138// pageCompressorPool reuses BestSpeed writers; each one owns ~600 KiB of
139// deflate state that would otherwise be allocated per compressed page.
140var pageCompressorPool = sync.Pool{New: func() any {
141 w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
142 if err != nil {
143 panic(err) // BestSpeed is a valid level
144 }
145 return w
146}}
147
148func (p *sealedPage) compressIfIdle() (bool, error) {
149 p.mu.Lock()
150 defer p.mu.Unlock()
151 if !p.observedOnce || p.observed != p.reads {
152 p.observed, p.observedOnce = p.reads, true
153 return false, nil
154 }
155 if p.frame.Height == 0 || p.incompressible {
156 return false, nil
157 }
158 if len(p.compressed) != 0 {
159 p.frame = renderer.Frame{}
160 return true, nil
161 }
162 // Encode the full backing, not an evicted-prefix wrapper. Synthetic IDs are
163 // local to this private envelope; public stable IDs remain on the wrappers.
164 chunk := &HistoryChunk{page: newSealedPage(p.frame), count: p.height, width: p.width,
165 bounds: make([]LineBound, p.height), rowIDs: make([]RowID, p.height)}
166 for y := range p.height {
167 chunk.bounds[y] = LineBound{End: p.width}
168 chunk.rowIDs[y] = RowID(y + 1)
169 }
170 raw, err := MarshalHistory(HistoryView{chunks: []*HistoryChunk{chunk}, rows: p.height, nextRowID: RowID(p.height + 1)})
171 if err != nil {
172 return false, err
173 }
174 if len(raw) > maxHistoryPageEncodedBytes {
175 return false, errors.New("history page exceeds encoded backing limit")
176 }
177 var out bytes.Buffer
178 w := pageCompressorPool.Get().(*zlib.Writer)
179 w.Reset(&out)
180 defer func() {
181 w.Reset(io.Discard)
182 pageCompressorPool.Put(w)
183 }()
184 if _, err := w.Write(raw); err != nil {
185 return false, err
186 }
187 if err := w.Close(); err != nil {
188 return false, err
189 }
190 if out.Len() >= len(raw) {
191 p.incompressible = true
192 return false, nil
193 }
194 p.compressed = make([]byte, out.Len())
195 copy(p.compressed, out.Bytes())
196 p.encodedSize = len(raw)
197 p.frame = renderer.Frame{}
198 return true, nil
199}
200
201// CompressIdle visits at most maxPages sealed pages. Call it from the history
202// owner's idle scheduler, never concurrently with append/eviction. A page must
203// remain unread across two visits; the newest sealed chunk and mutable tail are
204// kept hot, though unused mutable-tail capacity is released. There are no
205// internal goroutines, clocks, pooling or mmap mappings.
206func (h *History) CompressIdle(maxPages int) (int, error) {
207 if h == nil {
208 return 0, nil
209 }
210 h.trimIdleTail()
211 if maxPages <= 0 || len(h.chunks) < 2 {
212 return 0, nil
213 }
214 eligible := len(h.chunks) - 1
215 compressed := 0
216 for range min(maxPages, eligible) {
217 h.compressionCursor %= eligible
218 chunk := h.chunks[h.compressionCursor]
219 h.compressionCursor++
220 didCompress, err := chunk.page.compressIfIdle()
221 if err != nil {
222 return compressed, err
223 }
224 if didCompress {
225 compressed++
226 }
227 }
228 return compressed, nil
229}
230
231// HistoryCompressionStats describes retained physical backing, not process RSS.
232// ResidentLogicalBytes excludes Go allocator/map overhead; CompressedBytes is
233// the encoded backing size. Retention remains governed by LogicalBytes alone.
234type HistoryCompressionStats struct {
235 ColdPages int
236 CompressedBytes uint64
237 ResidentLogicalBytes uint64
238 Restores uint64
239}
240
241func (h *History) CompressionStats() HistoryCompressionStats {
242 var stats HistoryCompressionStats
243 if h == nil {
244 return stats
245 }
246 stats.ResidentLogicalBytes = h.tailBytes
247 for _, chunk := range h.chunks {
248 p := chunk.page
249 p.mu.Lock()
250 if p.frame.Height == 0 {
251 stats.ColdPages++
252 } else {
253 stats.ResidentLogicalBytes += p.frame.LogicalBytes()
254 }
255 stats.CompressedBytes += uint64(len(p.compressed))
256 stats.Restores += p.restores
257 p.mu.Unlock()
258 }
259 return stats
260}