X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/0821d23c64f4ffc856760e4914fa39ba2e3260e4..2532718d5288e1a2586fe5716cc41bb2a0210e16:/memory.go?ds=sidebyside

diff --git a/memory.go b/memory.go
index cb55cab..ca8277c 100644
--- a/memory.go
+++ b/memory.go
@@ -39,6 +39,10 @@ type Task struct {
 // start. It never builds the views again from the log, because a new view
 // differs from the live one and so loses the prompt cache.
 //
+// Each write goes to the operating system at once, and a syncer makes the
+// writes durable by group commit in the background. Flush waits until the
+// writes so far are durable.
+//
 // The summarizer builds the nodes in the queue in the background. The start
 // reads the indexes once to fill the queue again. After that, only a new
 // message or a new node adds work. Memory keeps the nodes of the views and the
@@ -48,6 +52,7 @@ type Memory struct {
 	mutex      sync.Mutex
 	root       string
 	log        *Log
+	syncer     *syncer
 	tree       *Tree
 	chat       *View
 	cview      *View
@@ -84,6 +89,11 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
 
 	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())
@@ -105,6 +115,7 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
 	}
 
 	m.mutex.Lock()
+	m.syncer.Written()
 	m.dispatch()
 	m.mutex.Unlock()
 
@@ -227,8 +238,8 @@ func (m *Memory) catchUp() error {
 	return m.save()
 }
 
-// Close stops the builds, waits for them, closes the files of the chat, and
-// releases its lock.
+// 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()
@@ -237,8 +248,12 @@ func (m *Memory) Close() error {
 	defer m.mutex.Unlock()
 
 	var err error
+	if m.syncer != nil {
+		err = m.syncer.Close()
+	}
+
 	if m.tree != nil {
-		err = m.tree.Close()
+		err = errors.Join(err, m.tree.Close())
 	}
 
 	return errors.Join(err, m.log.Close())
@@ -255,6 +270,8 @@ func (m *Memory) Append(kind, text string) (Message, error) {
 		return Message{}, err
 	}
 
+	defer m.syncer.Written()
+
 	err = m.addMessage(message)
 	if err != nil {
 		return Message{}, err
@@ -272,15 +289,24 @@ func (m *Memory) Append(kind, text string) (Message, error) {
 }
 
 // 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.
+// 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)
-	if free {
-		err := m.build(leaf)
+	switch {
+	case built:
+	case free:
+		err = m.build(leaf)
 		if err != nil {
 			return err
 		}
-	} else {
+	default:
 		m.queue.AddLeaf(message.ID)
 	}
 
@@ -448,6 +474,7 @@ func (m *Memory) run(key nodeKey, task Task) {
 			Text:  text,
 			Size:  len(text),
 		})
+		m.syncer.Written()
 	}
 
 	if err != nil {
@@ -481,6 +508,11 @@ func (m *Memory) Wait(ctx context.Context) error {
 	}
 }
 
+// Flush returns when every write before the call is durable.
+func (m *Memory) Flush(ctx context.Context) error {
+	return m.syncer.Flush(ctx)
+}
+
 // save writes the views to view.json.
 func (m *Memory) save() error {
 	data, err := json.Marshal(viewFile{