X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/f2d227db9c3b556c7b0e1cfe735ccc54311850e5..2124c559da8f164a43a3da987b399b6ac6125035:/memory.go?ds=sidebyside diff --git a/memory.go b/memory.go index 792fe21..a240b6c 100644 --- a/memory.go +++ b/memory.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "os" "path/filepath" "strings" @@ -12,24 +13,73 @@ import ( "time" ) -// errNoView tells that a chat has messages but no saved view. -var errNoView = errors.New("the chat has messages but no saved view") +var ( + // errMemoryNoView tells that a chat has messages but no saved view. + errMemoryNoView = errors.New("the chat has messages but no saved view") + + // errMemoryUnsettled tells that a line of the view stays pending, + // because its build failed. + errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message") +) + +// openMemory opens the chat in root and starts the builds that are ready. The +// clock gives the dates of new messages and the days of new nodes. logger gets +// one line for each failed build. The builds stop when ctx ends or at Close. +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 + } + + m := &Memory{ + root: root, + log: log, + queue: newBuildQueue(), + summarizer: summarizer, + logger: logger, + changed: make(chan struct{}), + } + + m.ctx, m.cancel = context.WithCancel(ctx) + + m.syncer, err = openSyncer(root) + if err != nil { + return nil, errors.Join(err, m.Close()) + } + + m.tree, err = openTree(ctx, root, log.Len(), clock) + if err != nil { + return nil, errors.Join(err, m.Close()) + } + + err = m.load() + if err != nil { + return nil, errors.Join(fmt.Errorf("load view: %w", err), m.Close()) + } + + err = m.restoreQueue() + if err != nil { + return nil, errors.Join(fmt.Errorf("restore build queue: %w", err), m.Close()) + } + + err = m.catchUp() + if err != nil { + return nil, errors.Join(err, m.Close()) + } + + m.mutex.Lock() + m.syncer.Written() + m.dispatch() + m.mutex.Unlock() + + return m, nil +} // Summarizer builds the text of one node with a model. type Summarizer interface { Summarize(ctx context.Context, task Task) (string, error) } -// Task is the build of the node (Level, Index). Context is the compaction view -// for the build. Input holds the source "kind: text" of a leaf, or the two -// child lines of a merge with their newlines as spaces. -type Task struct { - Level int - Index int - Context string - Input []string -} - // Memory is the memory of one chat: the message log, the summary tree, the chat // view, the compaction view, and the queue of the builds. The compaction view // is the chat view merged further, at the limits of 16 to 32 KB. It takes the @@ -52,106 +102,278 @@ type Memory struct { mutex sync.Mutex root string log *Log - syncer *syncer + syncer *Syncer tree *Tree chat *View cview *View - queue *queue + queue *BuildQueue summarizer Summarizer + logger *slog.Logger ctx context.Context cancel context.CancelFunc workers sync.WaitGroup changed chan struct{} } -// viewFile is the content of view.json. -type viewFile struct { - Chat viewState `json:"chat"` - Compaction viewState `json:"compaction"` +// Close stops the builds, waits for them, makes the writes durable, closes the +// files of the chat, and releases its lock. +func (m *Memory) Close() error { + m.cancel() + m.workers.Wait() + + m.mutex.Lock() + defer m.mutex.Unlock() + + var err error + + if m.syncer != nil { + err = m.syncer.Close() + } + + if m.tree != nil { + err = errors.Join(err, m.tree.Close()) + } + + 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, "") } -// openMemory opens the chat in root and starts the builds that are ready. The -// clock gives the dates of new messages and the days of new nodes. The builds -// stop when ctx ends or at Close. -func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer) (*Memory, error) { - log, err := openLog(ctx, root, clock) +// 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 nil, err + return Message{}, fmt.Errorf("append dated log message: %w", err) } - m := &Memory{ - root: root, - log: log, - queue: newQueue(), - summarizer: summarizer, - changed: make(chan struct{}), - } + defer m.syncer.Written() - m.ctx, m.cancel = context.WithCancel(ctx) + err = m.addMessage(message) + if err != nil { + return Message{}, fmt.Errorf("add dated message to memory: %w", err) + } - m.syncer, err = openSyncer(root) + err = m.save() if err != nil { - return nil, errors.Join(err, m.Close()) + return Message{}, fmt.Errorf("save dated message view: %w", err) } - m.tree, err = openTree(ctx, root, log.Len(), clock) + 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 nil, errors.Join(err, m.Close()) + return Message{}, fmt.Errorf("prepare queued message: %w", err) } - err = m.load() + message, err := m.appendDated(kind, text, "") if err != nil { - return nil, errors.Join(fmt.Errorf("load view: %w", err), m.Close()) + return Message{}, fmt.Errorf("append queued message: %w", err) } - err = m.restoreQueue() + 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 nil, errors.Join(fmt.Errorf("restore build queue: %w", err), m.Close()) + return fmt.Errorf("flush memory snapshot: %w", err) } - err = m.catchUp() + err = snapshot() if err != nil { - return nil, errors.Join(err, m.Close()) + 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() - m.syncer.Written() - m.dispatch() - m.mutex.Unlock() + defer m.mutex.Unlock() - return m, nil + 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()) if errors.Is(err, os.ErrNotExist) && m.log.Len() == 0 { - m.chat = newView(chatHigh, chatLow) - m.cview = newView(compactionHigh, compactionLow) + m.chat = newView(viewChatHigh, viewChatLow) + m.cview = newView(viewCompactionHigh, viewCompactionLow) return nil } if errors.Is(err, os.ErrNotExist) { - return errNoView + return errMemoryNoView } if err != nil { return fmt.Errorf("read view: %w", err) } - var file viewFile + var file MemoryViewFile + err = json.Unmarshal(data, &file) if err != nil { return fmt.Errorf("decode view: %w", err) } - m.chat, err = restoreView(file.Chat, chatHigh, chatLow, m.tree) + m.chat, err = restoreView(file.Chat, viewChatHigh, viewChatLow, m.tree) if err != nil { return fmt.Errorf("chat view: %w", err) } - m.cview, err = restoreView(file.Compaction, compactionHigh, compactionLow, m.tree) + m.cview, err = restoreView(file.Compaction, viewCompactionHigh, viewCompactionLow, m.tree) if err != nil { return fmt.Errorf("compaction view: %w", err) } @@ -188,6 +410,7 @@ func (m *Memory) restoreQueue() error { leaf, free := freeLeaf(message) if !free { m.queue.AddLeaf(id) + continue } @@ -238,54 +461,27 @@ func (m *Memory) catchUp() error { return m.save() } -// Close stops the builds, waits for them, makes the writes durable, closes the -// files of the chat, and releases its lock. -func (m *Memory) Close() error { - m.cancel() - m.workers.Wait() - - m.mutex.Lock() - defer m.mutex.Unlock() - - var err error - if m.syncer != nil { - err = m.syncer.Close() - } - - if m.tree != nil { - err = errors.Join(err, m.tree.Close()) - } - - 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) { - m.mutex.Lock() - defer m.mutex.Unlock() +// dispatch starts the builds that are ready, up to the limit. A build whose +// input cannot be read fails and waits for the next message. +func (m *Memory) dispatch() { + for m.ctx.Err() == nil { + key, ready := m.queue.Next() + if !ready { + return + } - message, err := m.log.Append(kind, text) - if err != nil { - return Message{}, err - } + task, err := m.task(key) + if err != nil { + m.failed(key, fmt.Errorf("read its input: %w", err)) + m.queue.Fail(key) - defer m.syncer.Written() + continue + } - err = m.addMessage(message) - if err != nil { - return Message{}, err - } + m.workers.Add(1) - err = m.save() - if err != nil { - return Message{}, err + go m.run(key, task) } - - m.queue.Retry() - m.dispatch() - - return message, nil } // addMessage builds the leaf of message when it needs no model call, or else @@ -327,6 +523,28 @@ func (m *Memory) addMessage(message Message) error { 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 { @@ -338,15 +556,15 @@ func (m *Memory) build(node Node) error { m.chat.Refresh(node) m.cview.Refresh(node) - return m.settle(nodeKey{node.Level, node.Index}) + 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. -func (m *Memory) settle(key nodeKey) error { +func (m *Memory) settle(key NodeKey) error { for { - parent := nodeKey{key.level + 1, key.index / 2} + parent := NodeKey{key.level + 1, key.index / 2} ready, err := m.ready(parent) if err != nil || !ready { return err @@ -360,6 +578,7 @@ func (m *Memory) settle(key nodeKey) error { node, free := freeMerge(left, right) if !free { m.queue.AddMerge(parent) + return nil } @@ -372,62 +591,9 @@ func (m *Memory) settle(key nodeKey) error { } } -// 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 - } - } - - built, err := m.tree.Built(key.level, key.index) - - return !built, err -} - -// 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 -} - -// dispatch starts the builds that are ready, up to the limit. A build whose -// input cannot be read fails and waits for the next message. -func (m *Memory) dispatch() { - for m.ctx.Err() == nil { - key, ready := m.queue.Next() - if !ready { - return - } - - task, err := m.task(key) - if err != nil { - m.queue.Fail(key) - continue - } - - m.workers.Add(1) - go m.run(key, task) - } -} - // 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) { +func (m *Memory) task(key NodeKey) (Task, error) { task := Task{ Level: key.level, Index: key.index, @@ -459,7 +625,7 @@ func (m *Memory) task(key nodeKey) (Task, error) { // 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) { +func (m *Memory) run(key NodeKey, task Task) { defer m.workers.Done() text, err := m.summarizer.Summarize(m.ctx, task) @@ -477,6 +643,10 @@ func (m *Memory) run(key nodeKey, task Task) { m.syncer.Written() } + if err != nil && m.ctx.Err() == nil { + m.failed(key, err) + } + if err != nil { m.queue.Fail(key) } else { @@ -484,55 +654,37 @@ func (m *Memory) run(key nodeKey, task Task) { } close(m.changed) + m.changed = make(chan struct{}) m.dispatch() } -// 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 - } +// 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) +} - select { - case <-changed: - case <-ctx.Done(): - return ctx.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 } } -} -// Flush returns when every write before the call is durable. -func (m *Memory) Flush(ctx context.Context) error { - return m.syncer.Flush(ctx) -} + built, err := m.tree.Built(key.level, key.index) -// save writes the views to view.json. -func (m *Memory) save() error { - data, err := json.Marshal(viewFile{ - Chat: m.chat.State(), - Compaction: m.cview.State(), - }) - if err != nil { - return fmt.Errorf("encode view: %w", err) - } + return !built, err +} - err = writeFileAtomic(m.viewPath(), append(data, '\n')) - if err != nil { - return fmt.Errorf("save view: %w", err) +func (m *Memory) context(level, index int) string { + end := index + if level > 0 { + end = nodeEnd(level, index) } - return nil -} - -func (m *Memory) viewPath() string { - return filepath.Join(m.root, "view.json") + return m.cview.Context(end) } // View returns the chat view as the model reads it. @@ -543,45 +695,102 @@ func (m *Memory) View() string { return m.chat.Render() } -// Len returns the number of messages in the log. -func (m *Memory) Len() int { +// Context returns the compaction view for the build of node (l, i). It ends +// before the message of a leaf, or after the last message of a merge, and it +// stops at the first line that is not built. +func (m *Memory) Context(level, index int) string { m.mutex.Lock() defer m.mutex.Unlock() - return m.log.Len() + return m.context(level, index) } -// Message reads the message with the given ID. -func (m *Memory) Message(id int) (Message, error) { +// rebuildComplete checks the whole tree, not only the visible view. A compactor +// can fail without an HTTP error, for example with an empty line. Such a failed +// merge can be absent from the view and must prevent a swap. +func (m *Memory) rebuildComplete() error { m.mutex.Lock() defer m.mutex.Unlock() - return m.log.Message(id) + if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 { + return errors.New("rebuild failed: summary tree has unfinished nodes") + } + + return nil } -// Node reads the node (l, i) if it is built. -func (m *Memory) Node(level, index int) (Node, bool, error) { +// buildModelLines returns the sizes of the nodes that the model wrote, in tree +// order. A node that fits its source with no model call is not one of them. +func (m *Memory) buildModelLines() ([]int, error) { m.mutex.Lock() defer m.mutex.Unlock() - return m.tree.Node(level, index) -} + keys, err := m.tree.Keys() + if err != nil { + return nil, fmt.Errorf("list built nodes: %w", err) + } -// Context returns the compaction view for the build of node (l, i). It ends -// before the message of a leaf, or after the last message of a merge, and it -// stops at the first line that is not built. -func (m *Memory) Context(level, index int) string { - m.mutex.Lock() - defer m.mutex.Unlock() + var sizes []int - return m.context(level, index) + for _, key := range keys { + var node Node + + node, _, err = m.tree.Node(key.level, key.index) + if err != nil { + return nil, fmt.Errorf("read built node: %w", err) + } + + var model bool + + model, err = m.buildBuiltByModel(key) + if err != nil { + return nil, fmt.Errorf("check model source: %w", err) + } + + if model { + sizes = append(sizes, node.Size) + } + } + + return sizes, nil } -func (m *Memory) context(level, index int) string { - end := index - if level > 0 { - end = nodeEnd(level, index) +// buildBuiltByModel tells if the node key needed a model call: a leaf whose +// source is over nodeBytes, or a merge whose children do not fit in one node. +func (m *Memory) buildBuiltByModel(key NodeKey) (bool, error) { + if key.level == 0 { + message, err := m.log.Message(key.index) + if err != nil { + return false, fmt.Errorf("read leaf source: %w", err) + } + + _, free := freeLeaf(message) + + return !free, nil } - return m.cview.Context(end) + left, right, err := m.children(key) + if err != nil { + return false, fmt.Errorf("read merge sources: %w", err) + } + + _, free := freeMerge(left, right) + + return !free, nil +} + +// Task is the build of the node (Level, Index). Context is the compaction view +// for the build. Input holds the source "kind: text" of a leaf, or the two +// child lines of a merge with their newlines as spaces. +type Task struct { + Level int + Index int + Context string + Input []string +} + +// MemoryViewFile is the content of view.json. +type MemoryViewFile struct { + Chat ViewState `json:"chat"` + Compaction ViewState `json:"compaction"` }