X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/2daf91497c2cc70d7156f7c0d99b1451933a818a..d28e3b2858db322899a81b59a30400c1acad2b42:/memory.go diff --git a/memory.go b/memory.go index 751a5dc..602b86c 100644 --- a/memory.go +++ b/memory.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "os" "path/filepath" "strings" @@ -25,7 +25,7 @@ var ( // 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 @@ -108,7 +108,7 @@ type Memory struct { cview *View queue *BuildQueue summarizer Summarizer - logger *log.Logger + logger *slog.Logger ctx context.Context cancel context.CancelFunc workers sync.WaitGroup @@ -125,6 +125,7 @@ func (m *Memory) Close() error { defer m.mutex.Unlock() var err error + if m.syncer != nil { err = m.syncer.Close() } @@ -139,24 +140,35 @@ func (m *Memory) Close() error { // 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() @@ -165,9 +177,48 @@ func (m *Memory) Append(kind, text string) (Message, error) { 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 { @@ -188,6 +239,7 @@ func (m *Memory) children(key NodeKey) (Node, Node, error) { func (m *Memory) Wait(ctx context.Context) error { for { m.mutex.Lock() + running := m.queue.Running() changed := m.changed m.mutex.Unlock() @@ -211,6 +263,7 @@ func (m *Memory) Wait(ctx context.Context) error { func (m *Memory) Settle(ctx context.Context) error { for { m.mutex.Lock() + pending := m.chat.Pending() running := m.queue.Running() changed := m.changed @@ -237,6 +290,26 @@ 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 { @@ -289,6 +362,7 @@ func (m *Memory) load() error { } var file MemoryViewFile + err = json.Unmarshal(data, &file) if err != nil { return fmt.Errorf("decode view: %w", err) @@ -336,6 +410,7 @@ func (m *Memory) restoreQueue() error { leaf, free := freeLeaf(message) if !free { m.queue.AddLeaf(id) + continue } @@ -404,6 +479,7 @@ func (m *Memory) dispatch() { } m.workers.Add(1) + go m.run(key, task) } } @@ -502,6 +578,7 @@ func (m *Memory) settle(key NodeKey) error { node, free := freeMerge(left, right) if !free { m.queue.AddMerge(parent) + return nil } @@ -530,6 +607,7 @@ func (m *Memory) task(key NodeKey) (Task, error) { } task.Input = []string{leafSource(message)} + task.Images = imageFiles(message.Text) return task, nil } @@ -577,13 +655,14 @@ func (m *Memory) run(key NodeKey, task Task) { } 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. @@ -627,14 +706,90 @@ func (m *Memory) Context(level, index int) string { 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. +// child lines of a merge with their newlines as spaces. Images holds the leaf +// image filenames and is empty for merges. type Task struct { Level int Index int Context string Input []string + Images []string } // MemoryViewFile is the content of view.json.