X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/80e348724041a694002fc38f57e8868b9b9a2be9..98d137c3e945a09f76e85bd18238558c668fdca4:/memory.go?ds=sidebyside diff --git a/memory.go b/memory.go index df99a95..97be6a1 100644 --- a/memory.go +++ b/memory.go @@ -86,7 +86,7 @@ type MemoryViewFile struct { func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer, logger *slog.Logger) (*Memory, error) { log, err := openLog(ctx, root, clock) if err != nil { - return nil, err + return nil, fmt.Errorf("open memory log: %w", err) } m := &Memory{ @@ -155,212 +155,6 @@ func (m *Memory) Close() error { return errors.Join(err, m.log.Close()) } -// Append logs a new message, adds it to the tree and the views, and starts the -// builds that are ready, also the failed ones. -func (m *Memory) Append(kind, text string) (Message, error) { - return m.AppendDated(kind, text, "") -} - -// AppendDated appends a message with its RFC 3339 date and builds its nodes. An -// empty date uses the clock, as Append does. -func (m *Memory) AppendDated(kind, text, date string) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.appendDated(kind, text, date) -} - -// appendDated appends under the memory lock and updates the views and builds. -func (m *Memory) appendDated(kind, text, date string) (Message, error) { - message, err := m.log.AppendDated(kind, text, date) - if err != nil { - return Message{}, fmt.Errorf("append dated log message: %w", err) - } - - defer m.syncer.Written() - - err = m.addMessage(message) - if err != nil { - return Message{}, fmt.Errorf("add dated message to memory: %w", err) - } - - err = m.save() - if err != nil { - return Message{}, fmt.Errorf("save dated message view: %w", err) - } - - m.queue.Retry() - m.dispatch() - - return message, nil -} - -// AppendOnce appends a queued message. prepare saves its log ID before the -// append, under the memory lock. A saved ID permits recovery after a stop -// between that save and the append, without a second log entry. prepare must -// not call Memory. The caller stops delivery after a failed append. -func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - if id != nil && *id < m.log.Len() { - message, err := m.log.Message(*id) - if err != nil { - return Message{}, fmt.Errorf("read queued message: %w", err) - } - - if message.Kind != kind || message.Text != text { - return Message{}, errors.New("queued message does not match its log entry") - } - - return message, nil - } - - if id != nil && *id != m.log.Len() { - return Message{}, errors.New("queued message is beyond the log") - } - - err := prepare(m.log.Len()) - if err != nil { - return Message{}, fmt.Errorf("prepare queued message: %w", err) - } - - message, err := m.appendDated(kind, text, "") - if err != nil { - return Message{}, fmt.Errorf("append queued message: %w", err) - } - - return message, nil -} - -// children reads the two children of the node key. -func (m *Memory) children(key NodeKey) (Node, Node, error) { - var children [2]Node - - for n := range children { - child, built, err := m.tree.Node(key.level-1, 2*key.index+n) - if err != nil { - return Node{}, Node{}, err - } - - if !built { - return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n) - } - - children[n] = child - } - - return children[0], children[1], nil -} - -// Wait returns when no build runs. The failed builds wait for the next message. -func (m *Memory) Wait(ctx context.Context) error { - for { - m.mutex.Lock() - - running := m.queue.Running() - changed := m.changed - m.mutex.Unlock() - - if running == 0 { - return nil - } - - select { - case <-changed: - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// Settle returns when the chat view has no pending line, so that every message -// so far is summarized. A pending line with no build that runs belongs to a -// failed build, which waits for the next message, so Settle returns -// errMemoryUnsettled in place of a wait that never ends. -func (m *Memory) Settle(ctx context.Context) error { - for { - m.mutex.Lock() - - pending := m.chat.Pending() - running := m.queue.Running() - changed := m.changed - m.mutex.Unlock() - - if !pending { - return nil - } - - if running == 0 { - return errMemoryUnsettled - } - - select { - case <-changed: - case <-ctx.Done(): - return ctx.Err() - } - } -} - -// Flush returns when every write before the call is durable. -func (m *Memory) Flush(ctx context.Context) error { - return m.syncer.Flush(ctx) -} - -// Snapshot flushes all memory writes and runs snapshot while no append or -// summary write can change the log, indexes, tree, or views. snapshot must not -// call Memory. -func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error { - m.mutex.Lock() - defer m.mutex.Unlock() - - err := m.Flush(ctx) - if err != nil { - return fmt.Errorf("flush memory snapshot: %w", err) - } - - err = snapshot() - if err != nil { - return fmt.Errorf("copy memory snapshot: %w", err) - } - - return nil -} - -// ViewLines returns the lines of the chat view, each with its newline, without -// the leaf lines of the messages in skip. -func (m *Memory) ViewLines(skip map[int]bool) []string { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.chat.LinesWithout(skip) -} - -// Len returns the number of messages in the log. -func (m *Memory) Len() int { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.log.Len() -} - -// Message reads the message with the given ID. -func (m *Memory) Message(id int) (Message, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.log.Message(id) -} - -// Node reads the node (l, i) if it is built. -func (m *Memory) Node(level, index int) (Node, bool, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - - return m.tree.Node(level, index) -} - // load reads the saved views. func (m *Memory) load() error { data, err := os.ReadFile(m.viewPath()) @@ -413,7 +207,7 @@ func (m *Memory) restoreQueue() error { for id := range m.chat.end { built, err := m.tree.Built(0, id) if err != nil { - return err + return fmt.Errorf("check restored leaf: %w", err) } if built { @@ -422,7 +216,7 @@ func (m *Memory) restoreQueue() error { message, err := m.log.Message(id) if err != nil { - return err + return fmt.Errorf("read restored message: %w", err) } leaf, free := freeLeaf(message) @@ -434,13 +228,13 @@ func (m *Memory) restoreQueue() error { err = m.build(leaf) if err != nil { - return err + return fmt.Errorf("build restored leaf: %w", err) } } keys, err := m.tree.Keys() if err != nil { - return err + return fmt.Errorf("read built nodes: %w", err) } for _, key := range keys { @@ -450,7 +244,7 @@ func (m *Memory) restoreQueue() error { err = m.settle(key) if err != nil { - return err + return fmt.Errorf("settle restored node: %w", err) } } @@ -467,16 +261,21 @@ func (m *Memory) catchUp() error { for m.chat.end < m.log.Len() { message, err := m.log.Message(m.chat.end) if err != nil { - return err + return fmt.Errorf("read pending message: %w", err) } err = m.addMessage(message) if err != nil { - return err + return fmt.Errorf("add pending message: %w", err) } } - return m.save() + err := m.save() + if err != nil { + return fmt.Errorf("save caught-up memory: %w", err) + } + + return nil } // dispatch starts the builds that are ready, up to the limit. A build whose @@ -502,81 +301,29 @@ func (m *Memory) dispatch() { } } -// addMessage builds the leaf of message when it needs no model call, or else -// adds the leaf to the queue. Then it appends the leaf to the views. A crash -// can stop an append after the leaf build and before the view save, so a leaf -// that is built already stays as it is, and restoreQueue settles its parent. -func (m *Memory) addMessage(message Message) error { - built, err := m.tree.Built(0, message.ID) - if err != nil { - return err - } - - leaf, free := freeLeaf(message) - switch { - case built: - case free: - err = m.build(leaf) - if err != nil { - return err - } - default: - m.queue.AddLeaf(message.ID) - } +func (m *Memory) viewPath() string { + return filepath.Join(m.root, "view.json") +} - merged, err := m.chat.Append(m.tree) +// build adds node to the tree, keeps it in the views when it is a pending line, +// and settles its parent. +func (m *Memory) build(node Node) error { + err := m.tree.Add(node) if err != nil { - return fmt.Errorf("append to chat view: %w", err) + return fmt.Errorf("add built node: %w", err) } - _, err = m.cview.Append(m.tree) - if err == nil && merged { - err = m.cview.Coarsen(m.chat, m.tree) - } + m.chat.Refresh(node) + m.cview.Refresh(node) + err = m.settle(NodeKey{node.Level, node.Index}) if err != nil { - return fmt.Errorf("append to compaction view: %w", err) + return fmt.Errorf("settle built node: %w", err) } return nil } -// save writes the views to view.json. -func (m *Memory) save() error { - data, err := json.Marshal(MemoryViewFile{ - Chat: m.chat.State(), - Compaction: m.cview.State(), - }) - if err != nil { - return fmt.Errorf("encode view: %w", err) - } - - err = writeFileAtomic(m.viewPath(), append(data, '\n')) - if err != nil { - return fmt.Errorf("save view: %w", err) - } - - return nil -} - -func (m *Memory) viewPath() string { - return filepath.Join(m.root, "view.json") -} - -// build adds node to the tree, keeps it in the views when it is a pending line, -// and settles its parent. -func (m *Memory) build(node Node) error { - err := m.tree.Add(node) - if err != nil { - return err - } - - m.chat.Refresh(node) - m.cview.Refresh(node) - - return m.settle(NodeKey{node.Level, node.Index}) -} - // settle looks at the parent of the built node key when both children are // built. It builds each parent that needs no model call, up the tree, and adds // the first other parent to the queue. @@ -585,12 +332,16 @@ func (m *Memory) settle(key NodeKey) error { parent := NodeKey{key.level + 1, key.index / 2} ready, err := m.ready(parent) if err != nil || !ready { - return err + if err != nil { + return fmt.Errorf("check parent readiness: %w", err) + } + + return nil } left, right, err := m.children(parent) if err != nil { - return err + return fmt.Errorf("read parent children: %w", err) } node, free := freeMerge(left, right) @@ -602,13 +353,70 @@ func (m *Memory) settle(key NodeKey) error { err = m.tree.Add(node) if err != nil { - return err + return fmt.Errorf("add merged node: %w", err) } key = parent } } +// addMessage builds the leaf of message when it needs no model call, or else +// adds the leaf to the queue. Then it appends the leaf to the views. A crash +// can stop an append after the leaf build and before the view save, so a leaf +// that is built already stays as it is, and restoreQueue settles its parent. +func (m *Memory) addMessage(message Message) error { + built, err := m.tree.Built(0, message.ID) + if err != nil { + return fmt.Errorf("check message leaf: %w", err) + } + + leaf, free := freeLeaf(message) + switch { + case built: + case free: + err = m.build(leaf) + if err != nil { + return fmt.Errorf("build message leaf: %w", err) + } + default: + m.queue.AddLeaf(message.ID) + } + + merged, err := m.chat.Append(m.tree) + if err != nil { + return fmt.Errorf("append to chat view: %w", err) + } + + _, err = m.cview.Append(m.tree) + if err == nil && merged { + err = m.cview.Coarsen(m.chat, m.tree) + } + + if err != nil { + return fmt.Errorf("append to compaction view: %w", err) + } + + return nil +} + +// save writes the views to view.json. +func (m *Memory) save() error { + data, err := json.Marshal(MemoryViewFile{ + Chat: m.chat.State(), + Compaction: m.cview.State(), + }) + if err != nil { + return fmt.Errorf("encode view: %w", err) + } + + err = writeFileAtomic(m.viewPath(), append(data, '\n')) + if err != nil { + return fmt.Errorf("save view: %w", err) + } + + return nil +} + // task returns the build task of the node key, with its input read from the log // or the tree. func (m *Memory) task(key NodeKey) (Task, error) { @@ -621,7 +429,7 @@ func (m *Memory) task(key NodeKey) (Task, error) { if key.level == 0 { message, err := m.log.Message(key.index) if err != nil { - return Task{}, err + return Task{}, fmt.Errorf("read leaf message: %w", err) } task.Input = []string{leafSource(message)} @@ -632,7 +440,7 @@ func (m *Memory) task(key NodeKey) (Task, error) { left, right, err := m.children(key) if err != nil { - return Task{}, err + return Task{}, fmt.Errorf("read task children: %w", err) } for _, child := range []Node{left, right} { @@ -642,6 +450,11 @@ func (m *Memory) task(key NodeKey) (Task, error) { return task, nil } +// failed logs the failed build of the node key with its cause err. +func (m *Memory) failed(key NodeKey, err error) { + m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err) +} + // run builds one node with the summarizer. A failed build stays in the queue // until the next message. func (m *Memory) run(key NodeKey, task Task) { @@ -659,6 +472,7 @@ func (m *Memory) run(key NodeKey, task Task) { Text: text, Size: len(text), }) + m.syncer.Written() } @@ -678,23 +492,45 @@ func (m *Memory) run(key NodeKey, task Task) { m.dispatch() } -// failed logs the failed build of the node key with its cause err. -func (m *Memory) failed(key NodeKey, err error) { - m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err) -} - // ready tells if both children of the node key are built and the node is not. func (m *Memory) ready(key NodeKey) (bool, error) { for _, index := range []int{2 * key.index, 2*key.index + 1} { built, err := m.tree.Built(key.level-1, index) if err != nil || !built { - return false, err + if err != nil { + return false, fmt.Errorf("check child readiness: %w", err) + } + + return false, nil } } built, err := m.tree.Built(key.level, key.index) + if err != nil { + return !built, fmt.Errorf("check parent readiness: %w", err) + } - return !built, err + return !built, nil +} + +// children reads the two children of the node key. +func (m *Memory) children(key NodeKey) (Node, Node, error) { + var children [2]Node + + for n := range children { + child, built, err := m.tree.Node(key.level-1, 2*key.index+n) + if err != nil { + return Node{}, Node{}, fmt.Errorf("read child node: %w", err) + } + + if !built { + return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n) + } + + children[n] = child + } + + return children[0], children[1], nil } func (m *Memory) context(level, index int) string { @@ -706,6 +542,192 @@ func (m *Memory) context(level, index int) string { return m.cview.Context(end) } +// Append logs a new message, adds it to the tree and the views, and starts the +// builds that are ready, also the failed ones. +func (m *Memory) Append(kind, text string) (Message, error) { + return m.AppendDated(kind, text, "") +} + +// AppendDated appends a message with its RFC 3339 date and builds its nodes. An +// empty date uses the clock, as Append does. +func (m *Memory) AppendDated(kind, text, date string) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.appendDated(kind, text, date) +} + +// appendDated appends under the memory lock and updates the views and builds. +func (m *Memory) appendDated(kind, text, date string) (Message, error) { + message, err := m.log.AppendDated(kind, text, date) + if err != nil { + return Message{}, fmt.Errorf("append dated log message: %w", err) + } + + defer m.syncer.Written() + + err = m.addMessage(message) + if err != nil { + return Message{}, fmt.Errorf("add dated message to memory: %w", err) + } + + err = m.save() + if err != nil { + return Message{}, fmt.Errorf("save dated message view: %w", err) + } + + m.queue.Retry() + m.dispatch() + + return message, nil +} + +// AppendOnce appends a queued message. prepare saves its log ID before the +// append, under the memory lock. A saved ID permits recovery after a stop +// between that save and the append, without a second log entry. prepare must +// not call Memory. The caller stops delivery after a failed append. +func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + if id != nil && *id < m.log.Len() { + message, err := m.log.Message(*id) + if err != nil { + return Message{}, fmt.Errorf("read queued message: %w", err) + } + + if message.Kind != kind || message.Text != text { + return Message{}, errors.New("queued message does not match its log entry") + } + + return message, nil + } + + if id != nil && *id != m.log.Len() { + return Message{}, errors.New("queued message is beyond the log") + } + + err := prepare(m.log.Len()) + if err != nil { + return Message{}, fmt.Errorf("prepare queued message: %w", err) + } + + message, err := m.appendDated(kind, text, "") + if err != nil { + return Message{}, fmt.Errorf("append queued message: %w", err) + } + + return message, nil +} + +// Wait returns when no build runs. The failed builds wait for the next message. +func (m *Memory) Wait(ctx context.Context) error { + for { + m.mutex.Lock() + + running := m.queue.Running() + changed := m.changed + m.mutex.Unlock() + + if running == 0 { + return nil + } + + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// Settle returns when the chat view has no pending line, so that every message +// so far is summarized. A pending line with no build that runs belongs to a +// failed build, which waits for the next message, so Settle returns +// errMemoryUnsettled in place of a wait that never ends. +func (m *Memory) Settle(ctx context.Context) error { + for { + m.mutex.Lock() + + pending := m.chat.Pending() + running := m.queue.Running() + changed := m.changed + m.mutex.Unlock() + + if !pending { + return nil + } + + if running == 0 { + return errMemoryUnsettled + } + + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// Snapshot flushes all memory writes and runs snapshot while no append or +// summary write can change the log, indexes, tree, or views. snapshot must not +// call Memory. +func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error { + m.mutex.Lock() + defer m.mutex.Unlock() + + err := m.Flush(ctx) + if err != nil { + return fmt.Errorf("flush memory snapshot: %w", err) + } + + err = snapshot() + if err != nil { + return fmt.Errorf("copy memory snapshot: %w", err) + } + + return nil +} + +// Flush returns when every write before the call is durable. +func (m *Memory) Flush(ctx context.Context) error { + return m.syncer.Flush(ctx) +} + +// ViewLines returns the lines of the chat view, each with its newline, without +// the leaf lines of the messages in skip. +func (m *Memory) ViewLines(skip map[int]bool) []string { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.chat.LinesWithout(skip) +} + +// Len returns the number of messages in the log. +func (m *Memory) Len() int { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.log.Len() +} + +// Message reads the message with the given ID. +func (m *Memory) Message(id int) (Message, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.log.Message(id) +} + +// Node reads the node (l, i) if it is built. +func (m *Memory) Node(level, index int) (Node, bool, error) { + m.mutex.Lock() + defer m.mutex.Unlock() + + return m.tree.Node(level, index) +} + // View returns the chat view as the model reads it. func (m *Memory) View() string { m.mutex.Lock()