X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/d4dc9a3b10a6092d18fb30cfb36526556aa33227..4c199e7d93b6a4529ed5956807911b5f6c5489a7:/memory_test.go
diff --git a/memory_test.go b/memory_test.go
index ddbb014..fe050c4 100644
--- a/memory_test.go
+++ b/memory_test.go
@@ -2,17 +2,15 @@ package main
import (
"context"
- "errors"
"fmt"
+ "log"
"maps"
"os"
"path/filepath"
"slices"
"strings"
- "sync"
"testing"
"time"
- "unicode/utf8"
"github.com/stretchr/testify/require"
)
@@ -20,11 +18,12 @@ import (
func TestMemoryRestartLoadsTheSameView(t *testing.T) {
root := t.TempDir()
clock := newTestClock()
- memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
memory.chat.high = 200
memory.chat.low = 100
+
memory.cview.high = 80
memory.cview.low = 40
for n := range 100 {
@@ -39,7 +38,7 @@ func TestMemoryRestartLoadsTheSameView(t *testing.T) {
require.Less(t, len(memory.chat.lines), 100)
require.NoError(t, memory.Close())
- memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
defer func() {
@@ -54,8 +53,35 @@ func TestMemoryRestartLoadsTheSameView(t *testing.T) {
require.Equal(t, saved, loaded)
}
+func TestMemoryOpensAfterACrashBeforeTheViewSave(t *testing.T) {
+ root := t.TempDir()
+ clock := newTestClock()
+ memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
+ require.NoError(t, err)
+
+ _, err = memory.Append("user", "first")
+ require.NoError(t, err)
+
+ saved, err := os.ReadFile(filepath.Join(root, "view.json"))
+ require.NoError(t, err)
+
+ _, err = memory.Append("user", "second")
+ require.NoError(t, err)
+
+ view := memory.View()
+ require.NoError(t, memory.Close())
+
+ // The log and the tree hold message 1, but the view does not.
+ require.NoError(t, os.WriteFile(filepath.Join(root, "view.json"), saved, 0o600))
+
+ memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
+ require.NoError(t, err)
+ require.Equal(t, view, memory.View())
+ require.NoError(t, memory.Close())
+}
+
func TestMemoryKeepsTheCompactionViewCoarser(t *testing.T) {
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
defer func() {
@@ -64,6 +90,7 @@ func TestMemoryKeepsTheCompactionViewCoarser(t *testing.T) {
memory.chat.high = 200
memory.chat.low = 100
+
memory.cview.high = 80
memory.cview.low = 40
for n := range 64 {
@@ -86,7 +113,7 @@ func TestMemoryKeepsTheCompactionViewCoarser(t *testing.T) {
func TestMemoryRunsAtMostEightBuilds(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.gate = make(chan struct{})
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
defer func() {
@@ -94,37 +121,112 @@ func TestMemoryRunsAtMostEightBuilds(t *testing.T) {
}()
for n := range 12 {
- _, err = memory.Append("echo", longText(n))
+ _, err = memory.Append("echo", memoryLongText(n))
require.NoError(t, err)
}
memory.mutex.Lock()
+
running := slices.Collect(maps.Keys(memory.queue.running))
memory.mutex.Unlock()
- slices.SortFunc(running, func(a, b nodeKey) int {
+ slices.SortFunc(running, func(a, b NodeKey) int {
return a.index - b.index
})
- require.Equal(t, []nodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}, {0, 4}, {0, 5}, {0, 6}, {0, 7}}, running)
+ require.Equal(t, []NodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}, {0, 4}, {0, 5}, {0, 6}, {0, 7}}, running)
require.Eventually(t, func() bool {
- return summarizer.Active() == maxBuilds
+ return summarizer.Active() == buildMax
}, 10*time.Second, time.Millisecond)
close(summarizer.gate)
require.NoError(t, memory.Wait(t.Context()))
- require.Equal(t, maxBuilds, summarizer.peak)
+ require.Equal(t, buildMax, summarizer.peak)
for level, count := range []int{12, 6, 3, 1} {
for index := range count {
- require.True(t, memory.tree.Built(level, index), "node (%d, %d)", level, index)
+ require.True(t, treeIsBuilt(t, memory.tree, level, index), "node (%d, %d)", level, index)
}
}
}
+func TestMemorySettleWaitsForThePendingLines(t *testing.T) {
+ summarizer := newFakeSummarizer()
+ summarizer.gate = make(chan struct{})
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, memory.Close())
+ }()
+
+ _, err = memory.Append("echo", memoryLongText(0))
+ require.NoError(t, err)
+ require.Contains(t, memory.View(), viewPending)
+
+ settled := make(chan error, 1)
+ go func() {
+ settled <- memory.Settle(t.Context())
+ }()
+
+ select {
+ case err = <-settled:
+ require.Fail(t, "Settle returned before the build", "error: %v", err)
+ case <-time.After(50 * time.Millisecond):
+ }
+
+ close(summarizer.gate)
+ require.NoError(t, <-settled)
+ require.NotContains(t, memory.View(), viewPending)
+}
+
+func TestMemorySettleFailsWhileABuildFailed(t *testing.T) {
+ summarizer := newFakeSummarizer()
+ summarizer.SetFailing(true)
+
+ logs := &SyncBuffer{}
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, log.New(logs, "", 0))
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, memory.Close())
+ }()
+
+ _, err = memory.Append("echo", memoryLongText(0))
+ require.NoError(t, err)
+ require.ErrorIs(t, memory.Settle(t.Context()), errMemoryUnsettled)
+ require.Equal(t, "summary failed, retry at next message: node (0, 0): fake failure\n", logs.String())
+
+ summarizer.SetFailing(false)
+
+ _, err = memory.Append("user", "retry")
+ require.NoError(t, err)
+ require.NoError(t, memory.Settle(t.Context()))
+}
+
+func TestMemorySettleEndsWithItsContext(t *testing.T) {
+ summarizer := newFakeSummarizer()
+ summarizer.gate = make(chan struct{})
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, memory.Close())
+ }()
+
+ _, err = memory.Append("echo", memoryLongText(0))
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+ defer cancel()
+
+ require.ErrorIs(t, memory.Settle(ctx), context.DeadlineExceeded)
+}
+
func TestMemoryStartsALeafAfterFewerThanEightUnbuilt(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
defer func() {
@@ -132,49 +234,50 @@ func TestMemoryStartsALeafAfterFewerThanEightUnbuilt(t *testing.T) {
}()
for n := range 9 {
- _, err = memory.Append("echo", longText(n))
+ _, err = memory.Append("echo", memoryLongText(n))
require.NoError(t, err)
}
require.NoError(t, memory.Wait(t.Context()))
+
for _, task := range summarizer.Tasks() {
require.Less(t, task.Index, 8)
}
- require.False(t, memory.tree.Built(0, 8))
+ require.False(t, treeIsBuilt(t, memory.tree, 0, 8))
summarizer.SetFailing(false)
_, err = memory.Append("user", "retry")
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
- require.True(t, memory.tree.Built(0, 8))
- require.True(t, memory.tree.Built(3, 0))
+ require.True(t, treeIsBuilt(t, memory.tree, 0, 8))
+ require.True(t, treeIsBuilt(t, memory.tree, 3, 0))
}
func TestMemoryMergesAfterBothHalves(t *testing.T) {
summarizer := newFakeSummarizer()
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
defer func() {
require.NoError(t, memory.Close())
}()
- _, err = memory.Append("echo", longText(0))
+ _, err = memory.Append("echo", memoryLongText(0))
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
- _, err = memory.Append("echo", longText(1))
+ _, err = memory.Append("echo", memoryLongText(1))
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
require.Len(t, summarizer.Tasks(), 3)
- left, _ := memory.tree.Node(0, 0)
- right, _ := memory.tree.Node(0, 1)
+ left := treeReadNode(t, memory.tree, 0, 0)
+ right := treeReadNode(t, memory.tree, 0, 1)
merge := summarizer.Tasks()[2]
require.Equal(t, []string{left.Text, right.Text}, merge.Input)
require.Equal(t, "\n0+1|"+left.Text+"\n1+1|"+right.Text+"\n", merge.Context)
- require.True(t, memory.tree.Built(1, 0))
+ require.True(t, treeIsBuilt(t, memory.tree, 1, 0))
}
func TestMemoryRestartRestoresTheQueue(t *testing.T) {
@@ -182,18 +285,19 @@ func TestMemoryRestartRestoresTheQueue(t *testing.T) {
clock := newTestClock()
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
- memory, err := openMemory(t.Context(), root, clock.Now, summarizer)
+
+ memory, err := openMemory(t.Context(), root, clock.Now, summarizer, discardLog())
require.NoError(t, err)
for n := range 4 {
- _, err = memory.Append("echo", longText(n))
+ _, err = memory.Append("echo", memoryLongText(n))
require.NoError(t, err)
}
require.NoError(t, memory.Wait(t.Context()))
require.NoError(t, memory.Close())
- memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
defer func() {
@@ -201,11 +305,11 @@ func TestMemoryRestartRestoresTheQueue(t *testing.T) {
}()
require.NoError(t, memory.Wait(t.Context()))
- require.True(t, memory.tree.Built(2, 0))
+ require.True(t, treeIsBuilt(t, memory.tree, 2, 0))
}
func TestMemoryBuildsFreeParents(t *testing.T) {
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
defer func() {
@@ -217,38 +321,41 @@ func TestMemoryBuildsFreeParents(t *testing.T) {
require.NoError(t, err)
}
- node, built := memory.tree.Node(2, 0)
- require.True(t, built)
+ node := treeReadNode(t, memory.tree, 2, 0)
require.Equal(t, "user: a\nuser: b\nuser: c\nuser: d", node.Text)
}
func TestMemoryKeepsALongMessageOutOfTheView(t *testing.T) {
summarizer := newFakeSummarizer()
summarizer.SetFailing(true)
- memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, discardLog())
require.NoError(t, err)
defer func() {
require.NoError(t, memory.Close())
}()
- _, err = memory.Append("echo", longText(0))
+ _, err = memory.Append("echo", memoryLongText(0))
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
- require.Equal(t, "\n0+1|"+pending+"\n", memory.View())
+ require.Equal(t, "\n0+1|"+viewPending+"\n", memory.View())
summarizer.SetFailing(false)
_, err = memory.Append("user", "again")
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
- require.Equal(t, "\n0+1|"+fakeLine(memory.task(nodeKey{0, 0}))+"\n1+1|user: again\n", memory.View())
+
+ task, err := memory.task(NodeKey{0, 0})
+ require.NoError(t, err)
+ require.Equal(t, "\n0+1|"+fakeLine(task)+"\n1+1|user: again\n", memory.View())
}
func TestMemoryCoversMessagesAfterTheSavedView(t *testing.T) {
root := t.TempDir()
clock := newTestClock()
- memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
_, err = memory.Append("user", "a")
@@ -258,7 +365,7 @@ func TestMemoryCoversMessagesAfterTheSavedView(t *testing.T) {
require.NoError(t, err)
require.NoError(t, memory.Close())
- memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
defer func() {
@@ -271,7 +378,7 @@ func TestMemoryCoversMessagesAfterTheSavedView(t *testing.T) {
func TestMemoryRefusesToRebuildAMissingView(t *testing.T) {
root := t.TempDir()
clock := newTestClock()
- memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
+ memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
_, err = memory.Append("user", "a")
@@ -279,94 +386,11 @@ func TestMemoryRefusesToRebuildAMissingView(t *testing.T) {
require.NoError(t, memory.Close())
require.NoError(t, os.Remove(filepath.Join(root, "view.json")))
- _, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer())
- require.ErrorIs(t, err, errNoView)
-}
-
-// fakeSummarizer builds each node with a line of fakeLine. It fails each task
-// while failing is true, and each task waits for gate when gate is not nil.
-type fakeSummarizer struct {
- mutex sync.Mutex
- gate chan struct{}
- failing bool
- active int
- peak int
- tasks []Task
+ _, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog())
+ require.ErrorIs(t, err, errMemoryNoView)
}
-func newFakeSummarizer() *fakeSummarizer {
- return &fakeSummarizer{}
-}
-
-func (f *fakeSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
- f.mutex.Lock()
- f.tasks = append(f.tasks, task)
- f.active++
- f.peak = max(f.peak, f.active)
- fail := f.failing
- f.mutex.Unlock()
-
- defer func() {
- f.mutex.Lock()
- f.active--
- f.mutex.Unlock()
- }()
-
- if f.gate != nil {
- select {
- case <-f.gate:
- case <-ctx.Done():
- return "", ctx.Err()
- }
- }
-
- if fail {
- return "", errors.New("fake failure")
- }
-
- return fakeLine(task), nil
-}
-
-// fakeLine returns a line of nodeBytes bytes or less that names the node and
-// starts with the input. A real line of this size makes each merge a model
-// call.
-func fakeLine(task Task) string {
- count := 1 << task.Level
- line := fmt.Sprintf("summary %d+%d: %s", task.Index*count, count, strings.Join(task.Input, " "))
- line = strings.ReplaceAll(line, "\n", " ")
- for len(line) > nodeBytes-12 {
- _, size := utf8.DecodeLastRuneInString(line)
- line = line[:len(line)-size]
- }
-
- return line
-}
-
-// longText returns a text that is too long for its own node.
-func longText(n int) string {
+// memoryLongText returns a text that is too long for its own node.
+func memoryLongText(n int) string {
return fmt.Sprintf("long %d ", n) + strings.Repeat("x", nodeBytes)
}
-
-// SetFailing makes the next tasks fail or not.
-func (f *fakeSummarizer) SetFailing(failing bool) {
- f.mutex.Lock()
- defer f.mutex.Unlock()
-
- f.failing = failing
-}
-
-// Active returns the number of tasks that run now.
-func (f *fakeSummarizer) Active() int {
- f.mutex.Lock()
- defer f.mutex.Unlock()
-
- return f.active
-}
-
-// Tasks returns the tasks so far.
-func (f *fakeSummarizer) Tasks() []Task {
- f.mutex.Lock()
- defer f.mutex.Unlock()
-
- return slices.Clone(f.tasks)
-}