]> git.giacomolli.uk - onechat/blob - memory.go
Complete context on forwarded local errors
[onechat] / memory.go
1 package main
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "log/slog"
9 "os"
10 "path/filepath"
11 "strings"
12 "sync"
13 "time"
14 )
15
16 var (
17 // errMemoryNoView tells that a chat has messages but no saved view.
18 errMemoryNoView = errors.New("the chat has messages but no saved view")
19
20 // errMemoryUnsettled tells that a line of the view stays pending,
21 // because its build failed.
22 errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message")
23 )
24
25 // Summarizer builds the text of one node with a model.
26 type Summarizer interface {
27 Summarize(ctx context.Context, task Task) (string, error)
28 }
29
30 // Memory is the memory of one chat: the message log, the summary tree, the chat
31 // view, the compaction view, and the queue of the builds. The compaction view
32 // is the chat view merged further, at the limits of 16 to 32 KB. It takes the
33 // same new lines, and it merges again in one batch when it is too large or when
34 // the chat view merges. So the compactions share its prefix in the prompt
35 // cache. It saves the views to view.json after each change and loads them at
36 // start. It never builds the views again from the log, because a new view
37 // differs from the live one and so loses the prompt cache.
38 //
39 // Each write goes to the operating system at once, and a syncer makes the
40 // writes durable by group commit in the background. Flush waits until the
41 // writes so far are durable.
42 //
43 // The summarizer builds the nodes in the queue in the background. The start
44 // reads the indexes once to fill the queue again. After that, only a new
45 // message or a new node adds work. Memory keeps the nodes of the views and the
46 // tasks of the running builds. It reads all other messages and nodes from their
47 // day files through the indexes.
48 type Memory struct {
49 mutex sync.Mutex
50 root string
51 log *Log
52 syncer *Syncer
53 tree *Tree
54 chat *View
55 cview *View
56 queue *BuildQueue
57 summarizer Summarizer
58 logger *slog.Logger
59 ctx context.Context
60 cancel context.CancelFunc
61 workers sync.WaitGroup
62 changed chan struct{}
63 }
64
65 // Task is the build of the node (Level, Index). Context is the compaction view
66 // for the build. Input holds the source "kind: text" of a leaf, or the two
67 // child lines of a merge with their newlines as spaces. Images holds the leaf
68 // image filenames and is empty for merges.
69 type Task struct {
70 Level int
71 Index int
72 Context string
73 Input []string
74 Images []string
75 }
76
77 // MemoryViewFile is the content of view.json.
78 type MemoryViewFile struct {
79 Chat ViewState `json:"chat"`
80 Compaction ViewState `json:"compaction"`
81 }
82
83 // openMemory opens the chat in root and starts the builds that are ready. The
84 // clock gives the dates of new messages and the days of new nodes. logger gets
85 // one line for each failed build. The builds stop when ctx ends or at Close.
86 func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer, logger *slog.Logger) (*Memory, error) {
87 log, err := openLog(ctx, root, clock)
88 if err != nil {
89 return nil, fmt.Errorf("open memory log: %w", err)
90 }
91
92 m := &Memory{
93 root: root,
94 log: log,
95 queue: newBuildQueue(),
96 summarizer: summarizer,
97 logger: logger,
98 changed: make(chan struct{}),
99 }
100
101 m.ctx, m.cancel = context.WithCancel(ctx)
102
103 m.syncer, err = openSyncer(root)
104 if err != nil {
105 return nil, errors.Join(err, m.Close())
106 }
107
108 m.tree, err = openTree(ctx, root, log.Len(), clock)
109 if err != nil {
110 return nil, errors.Join(err, m.Close())
111 }
112
113 err = m.load()
114 if err != nil {
115 return nil, errors.Join(fmt.Errorf("load view: %w", err), m.Close())
116 }
117
118 err = m.restoreQueue()
119 if err != nil {
120 return nil, errors.Join(fmt.Errorf("restore build queue: %w", err), m.Close())
121 }
122
123 err = m.catchUp()
124 if err != nil {
125 return nil, errors.Join(err, m.Close())
126 }
127
128 m.mutex.Lock()
129 m.syncer.Written()
130 m.dispatch()
131 m.mutex.Unlock()
132
133 return m, nil
134 }
135
136 // Close stops the builds, waits for them, makes the writes durable, closes the
137 // files of the chat, and releases its lock.
138 func (m *Memory) Close() error {
139 m.cancel()
140 m.workers.Wait()
141
142 m.mutex.Lock()
143 defer m.mutex.Unlock()
144
145 var err error
146
147 if m.syncer != nil {
148 err = m.syncer.Close()
149 }
150
151 if m.tree != nil {
152 err = errors.Join(err, m.tree.Close())
153 }
154
155 return errors.Join(err, m.log.Close())
156 }
157
158 // load reads the saved views.
159 func (m *Memory) load() error {
160 data, err := os.ReadFile(m.viewPath())
161 if errors.Is(err, os.ErrNotExist) && m.log.Len() == 0 {
162 m.chat = newView(viewChatHigh, viewChatLow)
163 m.cview = newView(viewCompactionHigh, viewCompactionLow)
164
165 return nil
166 }
167
168 if errors.Is(err, os.ErrNotExist) {
169 return errMemoryNoView
170 }
171
172 if err != nil {
173 return fmt.Errorf("read view: %w", err)
174 }
175
176 var file MemoryViewFile
177
178 err = json.Unmarshal(data, &file)
179 if err != nil {
180 return fmt.Errorf("decode view: %w", err)
181 }
182
183 m.chat, err = restoreView(file.Chat, viewChatHigh, viewChatLow, m.tree)
184 if err != nil {
185 return fmt.Errorf("chat view: %w", err)
186 }
187
188 m.cview, err = restoreView(file.Compaction, viewCompactionHigh, viewCompactionLow, m.tree)
189 if err != nil {
190 return fmt.Errorf("compaction view: %w", err)
191 }
192
193 if m.cview.end != m.chat.end {
194 return fmt.Errorf("compaction view covers %d messages, but the chat view %d", m.cview.end, m.chat.end)
195 }
196
197 if m.chat.end > m.log.Len() {
198 return fmt.Errorf("view covers %d messages, but the log has %d", m.chat.end, m.log.Len())
199 }
200
201 return nil
202 }
203
204 // restoreQueue adds the work of the messages in the view to the queue. It
205 // builds each node that needs no model call and that a crash left out.
206 func (m *Memory) restoreQueue() error {
207 for id := range m.chat.end {
208 built, err := m.tree.Built(0, id)
209 if err != nil {
210 return fmt.Errorf("check restored leaf: %w", err)
211 }
212
213 if built {
214 continue
215 }
216
217 message, err := m.log.Message(id)
218 if err != nil {
219 return fmt.Errorf("read restored message: %w", err)
220 }
221
222 leaf, free := freeLeaf(message)
223 if !free {
224 m.queue.AddLeaf(id)
225
226 continue
227 }
228
229 err = m.build(leaf)
230 if err != nil {
231 return fmt.Errorf("build restored leaf: %w", err)
232 }
233 }
234
235 keys, err := m.tree.Keys()
236 if err != nil {
237 return fmt.Errorf("read built nodes: %w", err)
238 }
239
240 for _, key := range keys {
241 if key.index%2 != 0 {
242 continue
243 }
244
245 err = m.settle(key)
246 if err != nil {
247 return fmt.Errorf("settle restored node: %w", err)
248 }
249 }
250
251 return nil
252 }
253
254 // catchUp adds the messages that the view does not cover yet. A crash can stop
255 // an append after the log write and before the view save.
256 func (m *Memory) catchUp() error {
257 if m.chat.end == m.log.Len() {
258 return nil
259 }
260
261 for m.chat.end < m.log.Len() {
262 message, err := m.log.Message(m.chat.end)
263 if err != nil {
264 return fmt.Errorf("read pending message: %w", err)
265 }
266
267 err = m.addMessage(message)
268 if err != nil {
269 return fmt.Errorf("add pending message: %w", err)
270 }
271 }
272
273 err := m.save()
274 if err != nil {
275 return fmt.Errorf("save caught-up memory: %w", err)
276 }
277
278 return nil
279 }
280
281 // dispatch starts the builds that are ready, up to the limit. A build whose
282 // input cannot be read fails and waits for the next message.
283 func (m *Memory) dispatch() {
284 for m.ctx.Err() == nil {
285 key, ready := m.queue.Next()
286 if !ready {
287 return
288 }
289
290 task, err := m.task(key)
291 if err != nil {
292 m.failed(key, fmt.Errorf("read its input: %w", err))
293 m.queue.Fail(key)
294
295 continue
296 }
297
298 m.workers.Add(1)
299
300 go m.run(key, task)
301 }
302 }
303
304 func (m *Memory) viewPath() string {
305 return filepath.Join(m.root, "view.json")
306 }
307
308 // build adds node to the tree, keeps it in the views when it is a pending line,
309 // and settles its parent.
310 func (m *Memory) build(node Node) error {
311 err := m.tree.Add(node)
312 if err != nil {
313 return fmt.Errorf("add built node: %w", err)
314 }
315
316 m.chat.Refresh(node)
317 m.cview.Refresh(node)
318
319 err = m.settle(NodeKey{node.Level, node.Index})
320 if err != nil {
321 return fmt.Errorf("settle built node: %w", err)
322 }
323
324 return nil
325 }
326
327 // settle looks at the parent of the built node key when both children are
328 // built. It builds each parent that needs no model call, up the tree, and adds
329 // the first other parent to the queue.
330 func (m *Memory) settle(key NodeKey) error {
331 for {
332 parent := NodeKey{key.level + 1, key.index / 2}
333 ready, err := m.ready(parent)
334 if err != nil || !ready {
335 if err != nil {
336 return fmt.Errorf("check parent readiness: %w", err)
337 }
338
339 return nil
340 }
341
342 left, right, err := m.children(parent)
343 if err != nil {
344 return fmt.Errorf("read parent children: %w", err)
345 }
346
347 node, free := freeMerge(left, right)
348 if !free {
349 m.queue.AddMerge(parent)
350
351 return nil
352 }
353
354 err = m.tree.Add(node)
355 if err != nil {
356 return fmt.Errorf("add merged node: %w", err)
357 }
358
359 key = parent
360 }
361 }
362
363 // addMessage builds the leaf of message when it needs no model call, or else
364 // adds the leaf to the queue. Then it appends the leaf to the views. A crash
365 // can stop an append after the leaf build and before the view save, so a leaf
366 // that is built already stays as it is, and restoreQueue settles its parent.
367 func (m *Memory) addMessage(message Message) error {
368 built, err := m.tree.Built(0, message.ID)
369 if err != nil {
370 return fmt.Errorf("check message leaf: %w", err)
371 }
372
373 leaf, free := freeLeaf(message)
374 switch {
375 case built:
376 case free:
377 err = m.build(leaf)
378 if err != nil {
379 return fmt.Errorf("build message leaf: %w", err)
380 }
381 default:
382 m.queue.AddLeaf(message.ID)
383 }
384
385 merged, err := m.chat.Append(m.tree)
386 if err != nil {
387 return fmt.Errorf("append to chat view: %w", err)
388 }
389
390 _, err = m.cview.Append(m.tree)
391 if err == nil && merged {
392 err = m.cview.Coarsen(m.chat, m.tree)
393 }
394
395 if err != nil {
396 return fmt.Errorf("append to compaction view: %w", err)
397 }
398
399 return nil
400 }
401
402 // save writes the views to view.json.
403 func (m *Memory) save() error {
404 data, err := json.Marshal(MemoryViewFile{
405 Chat: m.chat.State(),
406 Compaction: m.cview.State(),
407 })
408 if err != nil {
409 return fmt.Errorf("encode view: %w", err)
410 }
411
412 err = writeFileAtomic(m.viewPath(), append(data, '\n'))
413 if err != nil {
414 return fmt.Errorf("save view: %w", err)
415 }
416
417 return nil
418 }
419
420 // task returns the build task of the node key, with its input read from the log
421 // or the tree.
422 func (m *Memory) task(key NodeKey) (Task, error) {
423 task := Task{
424 Level: key.level,
425 Index: key.index,
426 Context: m.context(key.level, key.index),
427 }
428
429 if key.level == 0 {
430 message, err := m.log.Message(key.index)
431 if err != nil {
432 return Task{}, fmt.Errorf("read leaf message: %w", err)
433 }
434
435 task.Input = []string{leafSource(message)}
436 task.Images = imageFiles(message.Text)
437
438 return task, nil
439 }
440
441 left, right, err := m.children(key)
442 if err != nil {
443 return Task{}, fmt.Errorf("read task children: %w", err)
444 }
445
446 for _, child := range []Node{left, right} {
447 task.Input = append(task.Input, strings.ReplaceAll(child.Text, "\n", " "))
448 }
449
450 return task, nil
451 }
452
453 // failed logs the failed build of the node key with its cause err.
454 func (m *Memory) failed(key NodeKey, err error) {
455 m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err)
456 }
457
458 // run builds one node with the summarizer. A failed build stays in the queue
459 // until the next message.
460 func (m *Memory) run(key NodeKey, task Task) {
461 defer m.workers.Done()
462
463 text, err := m.summarizer.Summarize(m.ctx, task)
464
465 m.mutex.Lock()
466 defer m.mutex.Unlock()
467
468 if err == nil {
469 err = m.build(Node{
470 Level: key.level,
471 Index: key.index,
472 Text: text,
473 Size: len(text),
474 })
475
476 m.syncer.Written()
477 }
478
479 if err != nil && m.ctx.Err() == nil {
480 m.failed(key, err)
481 }
482
483 if err != nil {
484 m.queue.Fail(key)
485 } else {
486 m.queue.Done(key)
487 }
488
489 close(m.changed)
490
491 m.changed = make(chan struct{})
492 m.dispatch()
493 }
494
495 // ready tells if both children of the node key are built and the node is not.
496 func (m *Memory) ready(key NodeKey) (bool, error) {
497 for _, index := range []int{2 * key.index, 2*key.index + 1} {
498 built, err := m.tree.Built(key.level-1, index)
499 if err != nil || !built {
500 if err != nil {
501 return false, fmt.Errorf("check child readiness: %w", err)
502 }
503
504 return false, nil
505 }
506 }
507
508 built, err := m.tree.Built(key.level, key.index)
509 if err != nil {
510 return !built, fmt.Errorf("check parent readiness: %w", err)
511 }
512
513 return !built, nil
514 }
515
516 // children reads the two children of the node key.
517 func (m *Memory) children(key NodeKey) (Node, Node, error) {
518 var children [2]Node
519
520 for n := range children {
521 child, built, err := m.tree.Node(key.level-1, 2*key.index+n)
522 if err != nil {
523 return Node{}, Node{}, fmt.Errorf("read child node: %w", err)
524 }
525
526 if !built {
527 return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n)
528 }
529
530 children[n] = child
531 }
532
533 return children[0], children[1], nil
534 }
535
536 func (m *Memory) context(level, index int) string {
537 end := index
538 if level > 0 {
539 end = nodeEnd(level, index)
540 }
541
542 return m.cview.Context(end)
543 }
544
545 // Append logs a new message, adds it to the tree and the views, and starts the
546 // builds that are ready, also the failed ones.
547 func (m *Memory) Append(kind, text string) (Message, error) {
548 return m.AppendDated(kind, text, "")
549 }
550
551 // AppendDated appends a message with its RFC 3339 date and builds its nodes. An
552 // empty date uses the clock, as Append does.
553 func (m *Memory) AppendDated(kind, text, date string) (Message, error) {
554 m.mutex.Lock()
555 defer m.mutex.Unlock()
556
557 return m.appendDated(kind, text, date)
558 }
559
560 // appendDated appends under the memory lock and updates the views and builds.
561 func (m *Memory) appendDated(kind, text, date string) (Message, error) {
562 message, err := m.log.AppendDated(kind, text, date)
563 if err != nil {
564 return Message{}, fmt.Errorf("append dated log message: %w", err)
565 }
566
567 defer m.syncer.Written()
568
569 err = m.addMessage(message)
570 if err != nil {
571 return Message{}, fmt.Errorf("add dated message to memory: %w", err)
572 }
573
574 err = m.save()
575 if err != nil {
576 return Message{}, fmt.Errorf("save dated message view: %w", err)
577 }
578
579 m.queue.Retry()
580 m.dispatch()
581
582 return message, nil
583 }
584
585 // AppendOnce appends a queued message. prepare saves its log ID before the
586 // append, under the memory lock. A saved ID permits recovery after a stop
587 // between that save and the append, without a second log entry. prepare must
588 // not call Memory. The caller stops delivery after a failed append.
589 func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) {
590 m.mutex.Lock()
591 defer m.mutex.Unlock()
592
593 if id != nil && *id < m.log.Len() {
594 message, err := m.log.Message(*id)
595 if err != nil {
596 return Message{}, fmt.Errorf("read queued message: %w", err)
597 }
598
599 if message.Kind != kind || message.Text != text {
600 return Message{}, errors.New("queued message does not match its log entry")
601 }
602
603 return message, nil
604 }
605
606 if id != nil && *id != m.log.Len() {
607 return Message{}, errors.New("queued message is beyond the log")
608 }
609
610 err := prepare(m.log.Len())
611 if err != nil {
612 return Message{}, fmt.Errorf("prepare queued message: %w", err)
613 }
614
615 message, err := m.appendDated(kind, text, "")
616 if err != nil {
617 return Message{}, fmt.Errorf("append queued message: %w", err)
618 }
619
620 return message, nil
621 }
622
623 // Wait returns when no build runs. The failed builds wait for the next message.
624 func (m *Memory) Wait(ctx context.Context) error {
625 for {
626 m.mutex.Lock()
627
628 running := m.queue.Running()
629 changed := m.changed
630 m.mutex.Unlock()
631
632 if running == 0 {
633 return nil
634 }
635
636 select {
637 case <-changed:
638 case <-ctx.Done():
639 return ctx.Err()
640 }
641 }
642 }
643
644 // Settle returns when the chat view has no pending line, so that every message
645 // so far is summarized. A pending line with no build that runs belongs to a
646 // failed build, which waits for the next message, so Settle returns
647 // errMemoryUnsettled in place of a wait that never ends.
648 func (m *Memory) Settle(ctx context.Context) error {
649 for {
650 m.mutex.Lock()
651
652 pending := m.chat.Pending()
653 running := m.queue.Running()
654 changed := m.changed
655 m.mutex.Unlock()
656
657 if !pending {
658 return nil
659 }
660
661 if running == 0 {
662 return errMemoryUnsettled
663 }
664
665 select {
666 case <-changed:
667 case <-ctx.Done():
668 return ctx.Err()
669 }
670 }
671 }
672
673 // Snapshot flushes all memory writes and runs snapshot while no append or
674 // summary write can change the log, indexes, tree, or views. snapshot must not
675 // call Memory.
676 func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error {
677 m.mutex.Lock()
678 defer m.mutex.Unlock()
679
680 err := m.Flush(ctx)
681 if err != nil {
682 return fmt.Errorf("flush memory snapshot: %w", err)
683 }
684
685 err = snapshot()
686 if err != nil {
687 return fmt.Errorf("copy memory snapshot: %w", err)
688 }
689
690 return nil
691 }
692
693 // Flush returns when every write before the call is durable.
694 func (m *Memory) Flush(ctx context.Context) error {
695 return m.syncer.Flush(ctx)
696 }
697
698 // ViewLines returns the lines of the chat view, each with its newline, without
699 // the leaf lines of the messages in skip.
700 func (m *Memory) ViewLines(skip map[int]bool) []string {
701 m.mutex.Lock()
702 defer m.mutex.Unlock()
703
704 return m.chat.LinesWithout(skip)
705 }
706
707 // Len returns the number of messages in the log.
708 func (m *Memory) Len() int {
709 m.mutex.Lock()
710 defer m.mutex.Unlock()
711
712 return m.log.Len()
713 }
714
715 // Message reads the message with the given ID.
716 func (m *Memory) Message(id int) (Message, error) {
717 m.mutex.Lock()
718 defer m.mutex.Unlock()
719
720 return m.log.Message(id)
721 }
722
723 // Node reads the node (l, i) if it is built.
724 func (m *Memory) Node(level, index int) (Node, bool, error) {
725 m.mutex.Lock()
726 defer m.mutex.Unlock()
727
728 return m.tree.Node(level, index)
729 }
730
731 // View returns the chat view as the model reads it.
732 func (m *Memory) View() string {
733 m.mutex.Lock()
734 defer m.mutex.Unlock()
735
736 return m.chat.Render()
737 }
738
739 // Context returns the compaction view for the build of node (l, i). It ends
740 // before the message of a leaf, or after the last message of a merge, and it
741 // stops at the first line that is not built.
742 func (m *Memory) Context(level, index int) string {
743 m.mutex.Lock()
744 defer m.mutex.Unlock()
745
746 return m.context(level, index)
747 }
748
749 // rebuildComplete checks the whole tree, not only the visible view. A compactor
750 // can fail without an HTTP error, for example with an empty line. Such a failed
751 // merge can be absent from the view and must prevent a swap.
752 func (m *Memory) rebuildComplete() error {
753 m.mutex.Lock()
754 defer m.mutex.Unlock()
755
756 if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 {
757 return errors.New("rebuild failed: summary tree has unfinished nodes")
758 }
759
760 return nil
761 }
762
763 // buildModelLines returns the sizes of the nodes that the model wrote, in tree
764 // order. A node that fits its source with no model call is not one of them.
765 func (m *Memory) buildModelLines() ([]int, error) {
766 m.mutex.Lock()
767 defer m.mutex.Unlock()
768
769 keys, err := m.tree.Keys()
770 if err != nil {
771 return nil, fmt.Errorf("list built nodes: %w", err)
772 }
773
774 var sizes []int
775
776 for _, key := range keys {
777 var node Node
778
779 node, _, err = m.tree.Node(key.level, key.index)
780 if err != nil {
781 return nil, fmt.Errorf("read built node: %w", err)
782 }
783
784 var model bool
785
786 model, err = m.buildBuiltByModel(key)
787 if err != nil {
788 return nil, fmt.Errorf("check model source: %w", err)
789 }
790
791 if model {
792 sizes = append(sizes, node.Size)
793 }
794 }
795
796 return sizes, nil
797 }
798
799 // buildBuiltByModel tells if the node key needed a model call: a leaf whose
800 // source is over nodeBytes, or a merge whose children do not fit in one node.
801 func (m *Memory) buildBuiltByModel(key NodeKey) (bool, error) {
802 if key.level == 0 {
803 message, err := m.log.Message(key.index)
804 if err != nil {
805 return false, fmt.Errorf("read leaf source: %w", err)
806 }
807
808 _, free := freeLeaf(message)
809
810 return !free, nil
811 }
812
813 left, right, err := m.children(key)
814 if err != nil {
815 return false, fmt.Errorf("read merge sources: %w", err)
816 }
817
818 _, free := freeMerge(left, right)
819
820 return !free, nil
821 }