"encoding/json"
"errors"
"fmt"
- "log"
+ "log/slog"
"os"
"path/filepath"
"strings"
// 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) {
+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
cview *View
queue *BuildQueue
summarizer Summarizer
- logger *log.Logger
+ logger *slog.Logger
ctx context.Context
cancel context.CancelFunc
workers sync.WaitGroup
defer m.mutex.Unlock()
var err error
+
if m.syncer != nil {
err = m.syncer.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()
- message, err := m.log.Append(kind, text)
+ 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{}, err
+ return Message{}, fmt.Errorf("append dated log message: %w", err)
}
defer m.syncer.Written()
err = m.addMessage(message)
if err != nil {
- return Message{}, err
+ return Message{}, fmt.Errorf("add dated message to memory: %w", err)
}
err = m.save()
if err != nil {
- return Message{}, err
+ return Message{}, fmt.Errorf("save dated message view: %w", err)
}
m.queue.Retry()
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 {
func (m *Memory) Wait(ctx context.Context) error {
for {
m.mutex.Lock()
+
running := m.queue.Running()
changed := m.changed
m.mutex.Unlock()
func (m *Memory) Settle(ctx context.Context) error {
for {
m.mutex.Lock()
+
pending := m.chat.Pending()
running := m.queue.Running()
changed := m.changed
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 {
}
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
}
}
m.workers.Add(1)
+
go m.run(key, task)
}
}
node, free := freeMerge(left, right)
if !free {
m.queue.AddMerge(parent)
+
return nil
}
}
close(m.changed)
+
m.changed = make(chan struct{})
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.Printf("a summary of the chat failed, and it is tried again at the next message: node (%d, %d): %v", key.level, key.index, err)
+ 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.
return m.context(level, index)
}
+// 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()
+
+ if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 {
+ return errors.New("rebuild failed: summary tree has unfinished nodes")
+ }
+
+ return nil
+}
+
+// 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()
+
+ keys, err := m.tree.Keys()
+ if err != nil {
+ return nil, fmt.Errorf("list built nodes: %w", err)
+ }
+
+ var sizes []int
+
+ 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
+}
+
+// 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
+ }
+
+ 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.
projects / onechat / blobdiff