"encoding/json"
"errors"
"fmt"
+ "log"
"os"
"path/filepath"
"strings"
"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 *log.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
mutex sync.Mutex
root string
log *Log
- syncer *syncer
+ syncer *Syncer
tree *Tree
chat *View
cview *View
queue *BuildQueue
summarizer Summarizer
+ logger *log.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()
-// 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)
- if err != nil {
- return nil, err
+ m.mutex.Lock()
+ defer m.mutex.Unlock()
+
+ var err error
+
+ if m.syncer != nil {
+ err = m.syncer.Close()
}
- m := &Memory{
- root: root,
- log: log,
- queue: newBuildQueue(),
- summarizer: summarizer,
- changed: make(chan struct{}),
+ if m.tree != nil {
+ err = errors.Join(err, m.tree.Close())
}
- m.ctx, m.cancel = context.WithCancel(ctx)
+ return errors.Join(err, m.log.Close())
+}
- m.syncer, err = openSyncer(root)
+// 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()
+
+ message, err := m.log.AppendDated(kind, text, date)
if err != nil {
- return nil, errors.Join(err, m.Close())
+ return Message{}, fmt.Errorf("append dated log message: %w", err)
}
- m.tree, err = openTree(ctx, root, log.Len(), clock)
+ defer m.syncer.Written()
+
+ err = m.addMessage(message)
if err != nil {
- return nil, errors.Join(err, m.Close())
+ return Message{}, fmt.Errorf("add dated message to memory: %w", err)
}
- err = m.load()
+ err = m.save()
if err != nil {
- return nil, errors.Join(fmt.Errorf("load view: %w", err), m.Close())
+ return Message{}, fmt.Errorf("save dated message view: %w", err)
}
- err = m.restoreQueue()
- if err != nil {
- return nil, errors.Join(fmt.Errorf("restore build queue: %w", err), m.Close())
+ m.queue.Retry()
+ m.dispatch()
+
+ 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
}
- err = m.catchUp()
- if err != nil {
- return nil, errors.Join(err, m.Close())
+ 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)
+}
+
+// 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.
}
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)
leaf, free := freeLeaf(message)
if !free {
m.queue.AddLeaf(id)
+
continue
}
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
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 {
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
node, free := freeMerge(left, right)
if !free {
m.queue.AddMerge(parent)
+
return nil
}
}
}
-// 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,
// 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)
m.syncer.Written()
}
+ if err != nil && m.ctx.Err() == nil {
+ m.failed(key, err)
+ }
+
if err != nil {
m.queue.Fail(key)
} else {
}
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.Printf("summary failed, retry at next message: node (%d, %d): %v", key.level, key.index, 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.
return m.chat.Render()
}
-// 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 {
+// 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.chat.LinesWithout(skip)
+ return m.context(level, index)
}
-// Len returns the number of messages in the log.
-func (m *Memory) Len() int {
+// 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.Len()
+ if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 {
+ return errors.New("rebuild failed: summary tree has unfinished nodes")
+ }
+
+ return nil
}
-// Message reads the message with the given ID.
-func (m *Memory) Message(id int) (Message, 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.log.Message(id)
-}
+ keys, err := m.tree.Keys()
+ if err != nil {
+ return nil, fmt.Errorf("list built nodes: %w", err)
+ }
-// 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()
+ var sizes []int
- return m.tree.Node(level, index)
-}
+ for _, key := range keys {
+ var node Node
-// 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()
+ node, _, err = m.tree.Node(key.level, key.index)
+ if err != nil {
+ return nil, fmt.Errorf("read built node: %w", err)
+ }
- return m.context(level, index)
+ 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"`
}
projects / onechat / blobdiff