import (
"bufio"
"context"
- "encoding/json"
"errors"
"fmt"
+ "io"
"os"
"path/filepath"
+ "slices"
+ "strconv"
+ "strings"
)
+// position names one line of a day file: the day as YYYYMMDD, the byte offset
+// of the line, and its length with the newline.
+type position struct {
+ day uint32
+ offset uint64
+ length uint32
+}
+
// journal appends JSON lines to one file for each day in dir, named
-// YYYY-MM-DD.jsonl. It keeps the file of the last day open.
+// YYYY-MM-DD.jsonl, and reads them back by their position. It keeps the file of
+// the last day open for writes and each file that it read open for reads.
type journal struct {
- dir string
- file *os.File
- day string
+ dir string
+ file *os.File
+ day uint32
+ size uint64
+ readers map[uint32]*os.File
}
-// Write appends line to the file of day with one write to the operating system,
-// without a buffer. A crash of the process then loses nothing.
-func (j *journal) Write(day string, line []byte) error {
- if day != j.day {
- err := j.Close()
+// Write appends line to the file of day, YYYY-MM-DD, with one write to the
+// operating system, without a buffer. A crash of the process then loses
+// nothing. It returns the position of the line.
+func (j *journal) Write(day string, line []byte) (position, error) {
+ number, err := dayNumber(day)
+ if err != nil {
+ return position{}, err
+ }
+
+ if number != j.day {
+ err = j.closeWriter()
if err != nil {
- return err
+ return position{}, err
}
j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
- return fmt.Errorf("open day file: %w", err)
+ return position{}, fmt.Errorf("open day file: %w", err)
+ }
+
+ info, err := j.file.Stat()
+ if err != nil {
+ return position{}, fmt.Errorf("read day file size: %w", err)
}
- j.day = day
+ j.day = number
+ j.size = uint64(info.Size())
}
- _, err := j.file.Write(line)
+ _, err = j.file.Write(line)
if err != nil {
- return fmt.Errorf("write line: %w", err)
+ return position{}, fmt.Errorf("write line: %w", err)
}
- return nil
+ place := position{
+ day: number,
+ offset: j.size,
+ length: uint32(len(line)),
+ }
+ j.size += uint64(len(line))
+
+ return place, nil
}
-// Close closes the open day file.
-func (j *journal) Close() error {
+func (j *journal) closeWriter() error {
if j.file == nil {
return nil
}
err := j.file.Close()
j.file = nil
- j.day = ""
+ j.day = 0
if err != nil {
return fmt.Errorf("close day file: %w", err)
}
return nil
}
-// readLines decodes each line of the day files in dir, oldest day first, and
-// gives it to add in file order.
-func readLines[T any](ctx context.Context, dir string, add func(T) error) error {
- paths, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
+// Read returns the line at place without its newline.
+func (j *journal) Read(place position) ([]byte, error) {
+ reader, open := j.readers[place.day]
+ if !open {
+ var err error
+ reader, err = os.Open(filepath.Join(j.dir, dayName(place.day)+".jsonl"))
+ if err != nil {
+ return nil, fmt.Errorf("open day file: %w", err)
+ }
+
+ if j.readers == nil {
+ j.readers = make(map[uint32]*os.File)
+ }
+
+ j.readers[place.day] = reader
+ }
+
+ line := make([]byte, place.length)
+ _, err := reader.ReadAt(line, int64(place.offset))
+ if err != nil {
+ return nil, fmt.Errorf("read line of %s: %w", dayName(place.day), err)
+ }
+
+ if len(line) == 0 || line[len(line)-1] != '\n' {
+ return nil, fmt.Errorf("line of %s at %d does not end its record", dayName(place.day), place.offset)
+ }
+
+ return line[:len(line)-1], nil
+}
+
+// Close closes the open day files.
+func (j *journal) Close() error {
+ err := j.closeWriter()
+ for day, reader := range j.readers {
+ err = errors.Join(err, reader.Close())
+ delete(j.readers, day)
+ }
+
+ return err
+}
+
+// Scan gives each line of the day files from place on to add, with its
+// position, oldest day first. The zero position starts at the first line.
+func (j *journal) Scan(ctx context.Context, place position, add func([]byte, position) error) error {
+ paths, err := filepath.Glob(filepath.Join(j.dir, "*.jsonl"))
if err != nil {
return fmt.Errorf("list day files: %w", err)
}
+ slices.Sort(paths)
for _, path := range paths {
- err = readFile(ctx, path, add)
+ day, err := dayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
+ if err != nil {
+ return err
+ }
+
+ if day < place.day {
+ continue
+ }
+
+ offset := uint64(0)
+ if day == place.day {
+ offset = place.offset
+ }
+
+ err = scanFile(ctx, path, day, offset, add)
if err != nil {
return err
}
return nil
}
-func readFile[T any](ctx context.Context, path string, add func(T) error) (err error) {
+// scanFile gives each line of the day file at path from offset on to add.
+func scanFile(ctx context.Context, path string, day uint32, offset uint64, add func([]byte, position) error) (err error) {
file, err := os.Open(path)
if err != nil {
return fmt.Errorf("open day file: %w", err)
err = errors.Join(err, file.Close())
}()
- scanner := bufio.NewScanner(file)
- scanner.Buffer(nil, 1<<30)
- for number := 1; scanner.Scan(); number++ {
+ _, err = file.Seek(int64(offset), io.SeekStart)
+ if err != nil {
+ return fmt.Errorf("seek day file: %w", err)
+ }
+
+ reader := bufio.NewReader(file)
+ for {
if ctx.Err() != nil {
return ctx.Err()
}
- var value T
- err = json.Unmarshal(scanner.Bytes(), &value)
- if err == nil {
- err = add(value)
+ line, err := reader.ReadBytes('\n')
+ if errors.Is(err, io.EOF) && len(line) == 0 {
+ return nil
+ }
+
+ if errors.Is(err, io.EOF) {
+ return fmt.Errorf("%s ends in an incomplete line at %d", filepath.Base(path), offset)
+ }
+
+ if err != nil {
+ return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+ }
+
+ place := position{
+ day: day,
+ offset: offset,
+ length: uint32(len(line)),
}
+ err = add(line[:len(line)-1], place)
if err != nil {
- return fmt.Errorf("%s line %d: %w", filepath.Base(path), number, err)
+ return fmt.Errorf("%s at %d: %w", filepath.Base(path), offset, err)
}
+
+ offset += uint64(len(line))
}
+}
- err = scanner.Err()
- if err != nil {
- return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+// dayNumber returns the day YYYY-MM-DD as the number YYYYMMDD.
+func dayNumber(day string) (uint32, error) {
+ number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32)
+ if err != nil || len(day) != len("2006-01-02") {
+ return 0, fmt.Errorf("invalid day %q", day)
}
- return nil
+ return uint32(number), nil
+}
+
+// dayName returns the day number YYYYMMDD as YYYY-MM-DD.
+func dayName(day uint32) string {
+ return fmt.Sprintf("%04d-%02d-%02d", day/10000, day/100%100, day%100)
}
// exclusive lock on the lock file makes sure that only one process writes the
// log. After a failed write, the end of the file is not known, so the log
// refuses all writes until it is opened again. The caller serializes all calls.
+//
+// The log keeps no message in memory. The index index/main.idx holds the
+// position of message i in record i, so a read of a message is one read of the
+// index and one of its day file. At open the log indexes the lines after the
+// last indexed one, so a missing index is built again from the day files.
type Log struct {
- clock func() time.Time
- lock *os.File
- journal journal
- messages []Message
- failure error
+ clock func() time.Time
+ lock *os.File
+ journal journal
+ index *records
+ failure error
}
-// openLog locks the chat in root and loads all of its messages. The clock gives
-// the date of each new message.
+// openLog locks the chat in root and indexes its new lines. The clock gives the
+// date of each new message.
func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
- err := os.MkdirAll(filepath.Join(root, "main"), 0o700)
- if err != nil {
- return nil, fmt.Errorf("create message directory: %w", err)
+ for _, dir := range []string{"main", "index"} {
+ err := os.MkdirAll(filepath.Join(root, dir), 0o700)
+ if err != nil {
+ return nil, fmt.Errorf("create %s directory: %w", dir, err)
+ }
}
lock, err := lockFile(filepath.Join(root, "lock"))
},
}
- err = l.load(ctx)
+ l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes)
+ if err != nil {
+ return nil, errors.Join(err, l.Close())
+ }
+
+ err = l.catchUp(ctx)
if err != nil {
- return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close())
+ return nil, errors.Join(fmt.Errorf("index message log: %w", err), l.Close())
}
return l, nil
}
-func (l *Log) load(ctx context.Context) error {
- return readLines(ctx, l.journal.dir, func(message Message) error {
- if message.ID != len(l.messages) {
- return fmt.Errorf("expected message %d, found %d", len(l.messages), message.ID)
+// catchUp indexes the lines after the last indexed line.
+func (l *Log) catchUp(ctx context.Context) error {
+ start := position{}
+ if l.index.count > 0 {
+ record, _, err := l.index.Read(l.index.count - 1)
+ if err != nil {
+ return err
+ }
+
+ last := decodePosition(record)
+ start = position{
+ day: last.day,
+ offset: last.offset + uint64(last.length),
+ }
+ }
+
+ return l.journal.Scan(ctx, start, func(line []byte, place position) error {
+ var message Message
+ err := json.Unmarshal(line, &message)
+ if err != nil {
+ return err
+ }
+
+ if message.ID != l.index.count {
+ return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID)
}
- err := validateMessage(message)
+ err = validateMessage(message)
if err != nil {
return fmt.Errorf("message %d: %w", message.ID, err)
}
- l.messages = append(l.messages, message)
-
- return nil
+ return l.addIndex(place)
})
}
+func (l *Log) addIndex(place position) error {
+ record := make([]byte, positionBytes)
+ encodePosition(record, place)
+
+ return l.index.Append(record)
+}
+
// Close releases the lock. The log refuses all calls after Close.
func (l *Log) Close() error {
err := l.journal.Close()
+ if l.index != nil {
+ err = errors.Join(err, l.index.Close())
+ l.index = nil
+ }
+
if l.lock != nil {
err = errors.Join(err, l.lock.Close())
l.lock = nil
now := l.clock()
message := Message{
- ID: len(l.messages),
+ ID: l.index.count,
Kind: kind,
Text: text,
Size: len(text),
return Message{}, fmt.Errorf("encode message: %w", err)
}
- err = l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
+ place, err := l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
+ if err == nil {
+ err = l.addIndex(place)
+ }
+
if err != nil {
l.failure = err
return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
}
- l.messages = append(l.messages, message)
-
return message, nil
}
// Len returns the number of messages in the log.
func (l *Log) Len() int {
- return len(l.messages)
+ return l.index.count
}
-// Message returns the message with the given ID.
-func (l *Log) Message(id int) Message {
- return l.messages[id]
+// Message reads the message with the given ID from its day file.
+func (l *Log) Message(id int) (Message, error) {
+ record, found, err := l.index.Read(id)
+ if err != nil {
+ return Message{}, err
+ }
+
+ if !found {
+ return Message{}, fmt.Errorf("no message %d", id)
+ }
+
+ line, err := l.journal.Read(decodePosition(record))
+ if err != nil {
+ return Message{}, fmt.Errorf("read message %d: %w", id, err)
+ }
+
+ var message Message
+ err = json.Unmarshal(line, &message)
+ if err != nil {
+ return Message{}, fmt.Errorf("decode message %d: %w", id, err)
+ }
+
+ if message.ID != id {
+ return Message{}, fmt.Errorf("index of message %d points to message %d", id, message.ID)
+ }
+
+ return message, nil
}
// validateMessage makes sure that a message has a known kind, valid UTF-8 text,
require.NoError(t, log.Close())
}()
- require.Equal(t, written, log.messages)
+ require.Equal(t, len(written), log.Len())
+ for _, message := range written {
+ read, err := log.Message(message.ID)
+ require.NoError(t, err)
+ require.Equal(t, message, read)
+ }
message, err := log.Append("user", "after the reopen")
require.NoError(t, err)
require.Equal(t, 6, message.ID)
}
+func TestLogBuildsAMissingIndexAgain(t *testing.T) {
+ root := t.TempDir()
+ clock := newTestClock()
+ log, err := openLog(t.Context(), root, clock.Now)
+ require.NoError(t, err)
+
+ for _, text := range []string{"a", "b", "c"} {
+ _, err = log.Append("user", text)
+ require.NoError(t, err)
+
+ clock.now = clock.now.Add(24 * time.Hour)
+ }
+
+ require.NoError(t, log.Close())
+
+ // A missing index, and an index that a crash cut in a record.
+ for _, cut := range []func(string) error{os.Remove, truncateToOneAndAHalf} {
+ require.NoError(t, cut(filepath.Join(root, "index", "main.idx")))
+
+ log, err = openLog(t.Context(), root, clock.Now)
+ require.NoError(t, err)
+ require.Equal(t, 3, log.Len())
+
+ message, err := log.Message(2)
+ require.NoError(t, err)
+ require.Equal(t, "c", message.Text)
+ require.NoError(t, log.Close())
+ }
+}
+
func TestLogWritesOneFilePerDay(t *testing.T) {
root := t.TempDir()
clock := newTestClock()
require.ErrorContains(t, err, "expected message 0, found 1")
}
+// truncateToOneAndAHalf cuts the index at path in its second record.
+func truncateToOneAndAHalf(path string) error {
+ return os.Truncate(path, positionBytes+positionBytes/2)
+}
+
// testClock is a clock that tests set by hand.
type testClock struct {
now time.Time
// differs from the live one and so loses the prompt cache.
//
// The summarizer builds the nodes in the queue in the background. The start
-// reads the log and the tree once to fill the queue again. After that, only a
-// new message or a new node adds work.
+// 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
+// tasks of the running builds. It reads all other messages and nodes from their
+// day files through the indexes.
type Memory struct {
mutex sync.Mutex
root string
// builds each node that needs no model call and that a crash left out.
func (m *Memory) restoreQueue() error {
for id := range m.chat.end {
- if m.tree.Built(0, id) {
+ built, err := m.tree.Built(0, id)
+ if err != nil {
+ return err
+ }
+
+ if built {
continue
}
- leaf, free := freeLeaf(m.log.Message(id))
+ message, err := m.log.Message(id)
+ if err != nil {
+ return err
+ }
+
+ leaf, free := freeLeaf(message)
if !free {
m.queue.AddLeaf(id)
continue
}
- err := m.build(leaf)
+ err = m.build(leaf)
if err != nil {
return err
}
}
- for _, node := range m.tree.Nodes() {
- if node.Index%2 != 0 {
+ keys, err := m.tree.Keys()
+ if err != nil {
+ return err
+ }
+
+ for _, key := range keys {
+ if key.index%2 != 0 {
continue
}
- err := m.settle(node)
+ err = m.settle(key)
if err != nil {
return err
}
}
for m.chat.end < m.log.Len() {
- err := m.addMessage(m.log.Message(m.chat.end))
+ message, err := m.log.Message(m.chat.end)
+ if err != nil {
+ return err
+ }
+
+ err = m.addMessage(message)
if err != nil {
return err
}
m.queue.AddLeaf(message.ID)
}
- merged := m.chat.Append(m.tree)
- m.cview.Append(m.tree)
- if merged {
- m.cview.Coarsen(m.chat, m.tree)
+ merged, err := m.chat.Append(m.tree)
+ if err != nil {
+ return fmt.Errorf("append to chat view: %w", err)
+ }
+
+ _, err = m.cview.Append(m.tree)
+ if err == nil && merged {
+ err = m.cview.Coarsen(m.chat, m.tree)
+ }
+
+ if err != nil {
+ return fmt.Errorf("append to compaction view: %w", err)
}
return nil
}
-// build adds node to the tree and settles its parent.
+// 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 {
err := m.tree.Add(node)
if err != nil {
return err
}
- return m.settle(node)
+ m.chat.Refresh(node)
+ m.cview.Refresh(node)
+
+ return m.settle(nodeKey{node.Level, node.Index})
}
-// settle looks at the parent of node 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(node Node) error {
+// 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 {
for {
- sibling, built := m.tree.Node(node.Level, node.Index^1)
- if !built || m.tree.Built(node.Level+1, node.Index/2) {
- return nil
+ parent := nodeKey{key.level + 1, key.index / 2}
+ ready, err := m.ready(parent)
+ if err != nil || !ready {
+ return err
}
- left, right := node, sibling
- if node.Index%2 != 0 {
- left, right = sibling, node
+ left, right, err := m.children(parent)
+ if err != nil {
+ return err
}
- parent, free := freeMerge(left, right)
+ node, free := freeMerge(left, right)
if !free {
- m.queue.AddMerge(nodeKey{node.Level + 1, node.Index / 2})
+ m.queue.AddMerge(parent)
return nil
}
- err := m.tree.Add(parent)
+ err = m.tree.Add(node)
if err != nil {
return err
}
- node = parent
+ key = parent
}
}
-// dispatch starts the builds that are ready, up to the limit.
+// 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()
return
}
- task := m.task(key)
+ 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.
-func (m *Memory) task(key nodeKey) 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) {
task := Task{
Level: key.level,
Index: key.index,
}
if key.level == 0 {
- task.Input = []string{leafSource(m.log.Message(key.index))}
- return task
+ message, err := m.log.Message(key.index)
+ if err != nil {
+ return Task{}, err
+ }
+
+ task.Input = []string{leafSource(message)}
+
+ return task, nil
}
- for _, index := range []int{2 * key.index, 2*key.index + 1} {
- child, _ := m.tree.Node(key.level-1, index)
+ left, right, err := m.children(key)
+ if err != nil {
+ return Task{}, err
+ }
+
+ for _, child := range []Node{left, right} {
task.Input = append(task.Input, strings.ReplaceAll(child.Text, "\n", " "))
}
- return task
+ return task, nil
}
// run builds one node with the summarizer. A failed build stays in the queue
m.mutex.Lock()
defer m.mutex.Unlock()
- return m.chat.Render(m.tree)
+ return m.chat.Render()
}
// Context returns the compaction view for the build of node (l, i). It ends
end = nodeEnd(level, index)
}
- return m.cview.Context(m.tree, end)
+ return m.cview.Context(end)
}
// writeFileAtomic replaces the file at path with data through a temporary file
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, isBuilt(t, memory.tree, level, index), "node (%d, %d)", level, index)
}
}
}
require.Less(t, task.Index, 8)
}
- require.False(t, memory.tree.Built(0, 8))
+ require.False(t, isBuilt(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, isBuilt(t, memory.tree, 0, 8))
+ require.True(t, isBuilt(t, memory.tree, 3, 0))
}
func TestMemoryMergesAfterBothHalves(t *testing.T) {
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 := readNode(t, memory.tree, 0, 0)
+ right := readNode(t, memory.tree, 0, 1)
merge := summarizer.Tasks()[2]
require.Equal(t, []string{left.Text, right.Text}, merge.Input)
require.Equal(t, "<chat>\n0+1|"+left.Text+"\n1+1|"+right.Text+"\n</chat>", merge.Context)
- require.True(t, memory.tree.Built(1, 0))
+ require.True(t, isBuilt(t, memory.tree, 1, 0))
}
func TestMemoryRestartRestoresTheQueue(t *testing.T) {
}()
require.NoError(t, memory.Wait(t.Context()))
- require.True(t, memory.tree.Built(2, 0))
+ require.True(t, isBuilt(t, memory.tree, 2, 0))
}
func TestMemoryBuildsFreeParents(t *testing.T) {
require.NoError(t, err)
}
- node, built := memory.tree.Node(2, 0)
- require.True(t, built)
+ node := readNode(t, memory.tree, 2, 0)
require.Equal(t, "user: a\nuser: b\nuser: c\nuser: d", node.Text)
}
_, err = memory.Append("user", "again")
require.NoError(t, err)
require.NoError(t, memory.Wait(t.Context()))
- require.Equal(t, "<chat>\n0+1|"+fakeLine(memory.task(nodeKey{0, 0}))+"\n1+1|user: again\n</chat>", memory.View())
+
+ task, err := memory.task(nodeKey{0, 0})
+ require.NoError(t, err)
+ require.Equal(t, "<chat>\n0+1|"+fakeLine(task)+"\n1+1|user: again\n</chat>", memory.View())
}
func TestMemoryCoversMessagesAfterTheSavedView(t *testing.T) {
--- /dev/null
+package main
+
+import (
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+)
+
+// positionBytes is the size of an encoded position.
+const positionBytes = 16
+
+// records is a file of fixed-size records, numbered from 0. Record n starts at
+// byte n*size. A record that was never written reads as zeros. An index is data
+// that the day files can give again, so records cuts a partial record at the
+// end of the file.
+type records struct {
+ file *os.File
+ size int
+ count int
+}
+
+// openRecords opens or creates the record file at path with records of size
+// bytes.
+func openRecords(path string, size int) (*records, error) {
+ file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+ if err != nil {
+ return nil, fmt.Errorf("open index: %w", err)
+ }
+
+ info, err := file.Stat()
+ if err != nil {
+ return nil, errors.Join(fmt.Errorf("read index size: %w", err), file.Close())
+ }
+
+ r := &records{
+ file: file,
+ size: size,
+ count: int(info.Size()) / size,
+ }
+
+ if info.Size()%int64(size) != 0 {
+ err = file.Truncate(int64(r.count * size))
+ if err != nil {
+ return nil, errors.Join(fmt.Errorf("cut partial index record: %w", err), file.Close())
+ }
+ }
+
+ return r, nil
+}
+
+// Read returns record n, or false when the file does not reach it.
+func (r *records) Read(n int) ([]byte, bool, error) {
+ if n >= r.count {
+ return nil, false, nil
+ }
+
+ record := make([]byte, r.size)
+ _, err := r.file.ReadAt(record, int64(n*r.size))
+ if err != nil && !errors.Is(err, io.EOF) {
+ return nil, false, fmt.Errorf("read index record %d: %w", n, err)
+ }
+
+ return record, true, nil
+}
+
+// Write writes record n.
+func (r *records) Write(n int, record []byte) error {
+ _, err := r.file.WriteAt(record, int64(n*r.size))
+ if err != nil {
+ return fmt.Errorf("write index record %d: %w", n, err)
+ }
+
+ r.count = max(r.count, n+1)
+
+ return nil
+}
+
+// Append writes the record after the last one.
+func (r *records) Append(record []byte) error {
+ return r.Write(r.count, record)
+}
+
+// Close closes the file.
+func (r *records) Close() error {
+ return r.file.Close()
+}
+
+// encodePosition writes place into the first positionBytes bytes of record.
+func encodePosition(record []byte, place position) {
+ binary.LittleEndian.PutUint32(record[0:], place.day)
+ binary.LittleEndian.PutUint32(record[4:], place.length)
+ binary.LittleEndian.PutUint64(record[8:], place.offset)
+}
+
+// decodePosition reads a position from the first positionBytes bytes of record.
+func decodePosition(record []byte) position {
+ return position{
+ day: binary.LittleEndian.Uint32(record[0:]),
+ length: binary.LittleEndian.Uint32(record[4:]),
+ offset: binary.LittleEndian.Uint64(record[8:]),
+ }
+}
"math/rand/v2"
"os"
"path/filepath"
+ "runtime"
"strings"
"testing"
"time"
batches := 0
low := chatHigh
+ long := make([]bool, syntheticMessages)
for id := range syntheticMessages {
if id == syntheticMessages/2 {
memory = requireRestart(t, memory, root, clock)
clock.now = clock.now.Add(time.Minute)
kind, text := syntheticMessage(random, id)
+ long[id] = len(kind)+len(": ")+len(text) > nodeBytes
lines := viewLines(memory)
_, err = memory.Append(kind, text)
require.NoError(t, err)
- size, merged := requireViews(t, memory, lines)
+ size, merged := requireViews(t, memory, lines, long, id%1000 == 0)
if merged {
batches++
low = min(low, size)
require.Greater(t, batches, 0)
require.NoError(t, memory.Wait(t.Context()))
- t.Logf("%d messages, %d nodes, %d batches in %s", syntheticMessages, len(memory.tree.nodes), batches, time.Since(start).Round(time.Second))
- t.Logf("chat view: %d lines, %d bytes, lowest after a batch %d bytes", len(memory.chat.lines), memory.chat.Size(memory.tree), low)
- t.Logf("compaction view: %d lines, %d bytes", len(memory.cview.lines), memory.cview.Size(memory.tree))
+ t.Logf("%d messages, %d nodes, %d batches in %s", syntheticMessages, memory.tree.Count(), batches, time.Since(start).Round(time.Second))
+ t.Logf("chat view: %d lines, %d bytes, lowest after a batch %d bytes", len(memory.chat.lines), memory.chat.Size(), low)
+ t.Logf("compaction view: %d lines, %d bytes", len(memory.cview.lines), memory.cview.Size())
+
+ runtime.GC()
+
+ var stats runtime.MemStats
+ runtime.ReadMemStats(&stats)
+ t.Logf("heap in use: %d KB", stats.HeapInuse/1024)
}
// syntheticMessage returns the kind and the text of message id. Most messages
return head + strings.Repeat(words, max(1, (size-len(head))/len(words)))
}
-// requireViews checks the views after a new message. Each view is at most its
-// high limit, and a batch merges the chat view to its low limit. The leaf of
-// the new message can be built after the append, so the sizes count it as the
-// pending mark, as the append did. It returns the size of the chat view and if
-// a batch merged it, from the number of lines before the message.
-func requireViews(t *testing.T, memory *Memory, lines int) (int, bool) {
+// requireViews checks the views after a new message, and with full compares
+// them with the tree. Each view is at most its high limit, and a batch merges
+// the chat view to its low limit. The leaf of the new message can be built
+// after the append, so the sizes count it as the pending mark, as the append
+// did. It returns the size of the chat view and if a batch merged it, from the
+// number of lines before the message.
+func requireViews(t *testing.T, memory *Memory, lines int, long []bool, full bool) (int, bool) {
t.Helper()
memory.mutex.Lock()
defer memory.mutex.Unlock()
- size := appendSize(memory.chat, memory.tree)
+ size := appendSize(memory.chat)
require.LessOrEqual(t, size, chatHigh)
- require.LessOrEqual(t, appendSize(memory.cview, memory.tree), compactionHigh)
+ require.LessOrEqual(t, appendSize(memory.cview), compactionHigh)
merged := len(memory.chat.lines) < lines
if merged {
require.LessOrEqual(t, size, chatLow)
}
- requireOnlySummaries(t, memory)
+ requireOnlySummaries(t, memory, long, full)
return size, merged
}
// requireOnlySummaries checks each line of the chat view as the model reads it.
-// A built line holds its node, a long message has a summary and not its text,
-// and an unbuilt line holds only the pending mark. The caller holds the lock of
-// memory.
-func requireOnlySummaries(t *testing.T, memory *Memory) {
+// A built line holds its node, a long message has a summary of the fake
+// summarizer and not its text, and an unbuilt line holds only the pending mark.
+// With full, it also checks that the view keeps the node of each built line.
+// The caller holds the lock of memory.
+func requireOnlySummaries(t *testing.T, memory *Memory, long []bool, full bool) {
t.Helper()
- rendered := strings.Split(memory.chat.Render(memory.tree), "\n")
+ rendered := strings.Split(memory.chat.Render(), "\n")
require.Len(t, rendered, len(memory.chat.lines)+2)
for n, line := range memory.chat.lines {
head := fmt.Sprintf("%d+%d|", line.index<<line.level, 1<<line.level)
- node, built := memory.tree.Node(line.level, line.index)
+ node, built := memory.chat.nodes[line]
+ if full {
+ require.Equal(t, isBuilt(t, memory.tree, line.level, line.index), built)
+ }
+
if !built {
require.Equal(t, 0, line.level)
require.Equal(t, head+pending, rendered[n+1])
}
require.Equal(t, head+strings.ReplaceAll(node.Text, "\n", " "), rendered[n+1])
- if line.level == 0 {
- source := leafSource(memory.log.Message(line.index))
- if len(source) > nodeBytes {
- require.NotEqual(t, source, node.Text, "message %d", line.index)
- }
+ if line.level == 0 && long[line.index] {
+ require.True(t, strings.HasPrefix(node.Text, "summary "), "message %d", line.index)
}
}
}
// appendSize returns the size of view with its last line as the pending mark.
-func appendSize(view *View, tree nodeReader) int {
+func appendSize(view *View) int {
last := view.lines[len(view.lines)-1]
- return view.Size(tree) - lineSize(tree, last) + len(pending)
+ return view.Size() - view.lineSize(last) + len(pending)
}
// viewLines returns the number of lines in the chat view of memory.
import (
"cmp"
"context"
+ "encoding/binary"
"encoding/json"
+ "errors"
"fmt"
"math/bits"
"os"
index int
}
+// treeRecordBytes is the size of a record of index/tree.idx: the position of
+// the node line, the node index, and the node level.
+const treeRecordBytes = 32
+
// Tree holds the built nodes of one chat. It writes one JSON line for each node
-// to tree/YYYY-MM-DD.jsonl, named for the local day of the build, and loads all
-// lines again at start. Each node is built once and never again. A node of a
-// higher level is built only after both of its children. After a failed write,
-// the tree refuses all writes until it is opened again. The caller holds the
-// lock of the chat and serializes all calls.
+// to tree/YYYY-MM-DD.jsonl, named for the local day of the build. Each node is
+// built once and never again. A node of a higher level is built only after both
+// of its children. After a failed write, the tree refuses all writes until it
+// is opened again. The caller holds the lock of the chat and serializes all
+// calls.
+//
+// The tree keeps no node in memory. The index index/tree.idx lists the node
+// lines in file order, and index/tree-L.idx holds the position of node (L, i)
+// in record i, with a zero record for a node that is not built. At open the
+// tree indexes the lines after the last indexed one, so a missing index is
+// built again from the day files.
type Tree struct {
clock func() time.Time
journal journal
- nodes map[nodeKey]Node
+ dir string
+ order *records
+ levels map[int]*records
failure error
}
-// openTree loads the nodes of the chat in root. A chat of total messages cannot
-// have a node that covers a later message.
+// openTree opens the tree of the chat in root and indexes its new lines. A chat
+// of total messages cannot have a node that covers a later message.
func openTree(ctx context.Context, root string, total int, clock func() time.Time) (*Tree, error) {
- err := os.MkdirAll(filepath.Join(root, "tree"), 0o700)
- if err != nil {
- return nil, fmt.Errorf("create node directory: %w", err)
+ for _, dir := range []string{"tree", "index"} {
+ err := os.MkdirAll(filepath.Join(root, dir), 0o700)
+ if err != nil {
+ return nil, fmt.Errorf("create %s directory: %w", dir, err)
+ }
}
t := &Tree{
journal: journal{
dir: filepath.Join(root, "tree"),
},
- nodes: make(map[nodeKey]Node),
+ dir: filepath.Join(root, "index"),
+ levels: make(map[int]*records),
+ }
+
+ err := t.open(ctx, total)
+ if err != nil {
+ return nil, errors.Join(fmt.Errorf("index tree: %w", err), t.Close())
+ }
+
+ return t, nil
+}
+
+func (t *Tree) open(ctx context.Context, total int) error {
+ var err error
+ t.order, err = openRecords(filepath.Join(t.dir, "tree.idx"), treeRecordBytes)
+ if err != nil {
+ return err
+ }
+
+ start := position{}
+ if t.order.count == 0 {
+ err = t.removeLevels()
+ if err != nil {
+ return err
+ }
+ } else {
+ record, _, err := t.order.Read(t.order.count - 1)
+ if err != nil {
+ return err
+ }
+
+ // A crash can stop an add between the two index writes, so the
+ // last node gets its level record again.
+ node, last := decodeTreeRecord(record)
+ err = t.writeLevel(node, last)
+ if err != nil {
+ return err
+ }
+
+ start = position{
+ day: last.day,
+ offset: last.offset + uint64(last.length),
+ }
}
- err = readLines(ctx, t.journal.dir, func(node Node) error {
- err := t.check(node)
+ return t.journal.Scan(ctx, start, func(line []byte, place position) error {
+ var node Node
+ err := json.Unmarshal(line, &node)
+ if err != nil {
+ return err
+ }
+
+ err = t.check(node)
if err != nil {
return err
}
return fmt.Errorf("node (%d, %d) covers messages after the log", node.Level, node.Index)
}
- t.nodes[nodeKey{node.Level, node.Index}] = node
-
- return nil
+ return t.addIndex(nodeKey{node.Level, node.Index}, place)
})
+}
+
+// removeLevels removes the level indexes, which are not valid without the order
+// index.
+func (t *Tree) removeLevels() error {
+ paths, err := filepath.Glob(filepath.Join(t.dir, "tree-*.idx"))
if err != nil {
- return nil, fmt.Errorf("load tree: %w", err)
+ return fmt.Errorf("list level indexes: %w", err)
}
- return t, nil
+ for _, path := range paths {
+ err = os.Remove(path)
+ if err != nil {
+ return fmt.Errorf("remove level index: %w", err)
+ }
+ }
+
+ return nil
+}
+
+// writeLevel writes the position of node to its level index.
+func (t *Tree) writeLevel(node nodeKey, place position) error {
+ level, err := t.level(node.level)
+ if err != nil {
+ return err
+ }
+
+ record := make([]byte, positionBytes)
+ encodePosition(record, place)
+
+ return level.Write(node.index, record)
+}
+
+// level returns the index of level, and opens it at first use.
+func (t *Tree) level(level int) (*records, error) {
+ index, open := t.levels[level]
+ if open {
+ return index, nil
+ }
+
+ index, err := openRecords(filepath.Join(t.dir, fmt.Sprintf("tree-%d.idx", level)), positionBytes)
+ if err != nil {
+ return nil, err
+ }
+
+ t.levels[level] = index
+
+ return index, nil
}
// check makes sure that node is valid and can join the tree now.
return fmt.Errorf("node (%d, %d) has no valid text of its size", node.Level, node.Index)
}
- if t.Built(node.Level, node.Index) {
+ built, err := t.Built(node.Level, node.Index)
+ if err != nil {
+ return err
+ }
+
+ if built {
return fmt.Errorf("node (%d, %d) is already built", node.Level, node.Index)
}
- if node.Level > 0 && (!t.Built(node.Level-1, 2*node.Index) || !t.Built(node.Level-1, 2*node.Index+1)) {
- return fmt.Errorf("node (%d, %d) has a child that is not built", node.Level, node.Index)
+ if node.Level == 0 {
+ return nil
+ }
+
+ for _, index := range []int{2 * node.Index, 2*node.Index + 1} {
+ built, err = t.Built(node.Level-1, index)
+ if err != nil {
+ return err
+ }
+
+ if !built {
+ return fmt.Errorf("node (%d, %d) has a child that is not built", node.Level, node.Index)
+ }
}
return nil
}
-// Close closes the open day file.
+// addIndex adds the node line at place to both indexes.
+func (t *Tree) addIndex(node nodeKey, place position) error {
+ record := make([]byte, treeRecordBytes)
+ encodePosition(record, place)
+ binary.LittleEndian.PutUint64(record[positionBytes:], uint64(node.index))
+ binary.LittleEndian.PutUint32(record[positionBytes+8:], uint32(node.level))
+
+ err := t.order.Append(record)
+ if err != nil {
+ return err
+ }
+
+ return t.writeLevel(node, place)
+}
+
+// Close closes the files of the tree.
func (t *Tree) Close() error {
- return t.journal.Close()
+ err := t.journal.Close()
+ if t.order != nil {
+ err = errors.Join(err, t.order.Close())
+ t.order = nil
+ }
+
+ for level, index := range t.levels {
+ err = errors.Join(err, index.Close())
+ delete(t.levels, level)
+ }
+
+ return err
}
// Add writes a new node to the tree.
return fmt.Errorf("encode node: %w", err)
}
- err = t.journal.Write(t.clock().Format(time.DateOnly), append(line, '\n'))
+ place, err := t.journal.Write(t.clock().Format(time.DateOnly), append(line, '\n'))
+ if err == nil {
+ err = t.addIndex(nodeKey{node.Level, node.Index}, place)
+ }
+
if err != nil {
t.failure = err
return fmt.Errorf("write node (%d, %d): %w", node.Level, node.Index, err)
}
- t.nodes[nodeKey{node.Level, node.Index}] = node
-
return nil
}
// Built tells if the node (l, i) is built.
-func (t *Tree) Built(level, index int) bool {
- _, built := t.nodes[nodeKey{level, index}]
- return built
+func (t *Tree) Built(level, index int) (bool, error) {
+ _, built, err := t.place(level, index)
+ return built, err
+}
+
+// place returns the position of the node (l, i) if it is built.
+func (t *Tree) place(level, index int) (position, bool, error) {
+ if level < 0 || level >= bits.UintSize-2 || index < 0 {
+ return position{}, false, nil
+ }
+
+ records, err := t.level(level)
+ if err != nil {
+ return position{}, false, err
+ }
+
+ record, found, err := records.Read(index)
+ if err != nil || !found {
+ return position{}, false, err
+ }
+
+ place := decodePosition(record)
+
+ return place, place.length > 0, nil
+}
+
+// Node reads the node (l, i) if it is built.
+func (t *Tree) Node(level, index int) (Node, bool, error) {
+ place, built, err := t.place(level, index)
+ if err != nil || !built {
+ return Node{}, false, err
+ }
+
+ line, err := t.journal.Read(place)
+ if err != nil {
+ return Node{}, false, fmt.Errorf("read node (%d, %d): %w", level, index, err)
+ }
+
+ var node Node
+ err = json.Unmarshal(line, &node)
+ if err != nil {
+ return Node{}, false, fmt.Errorf("decode node (%d, %d): %w", level, index, err)
+ }
+
+ if node.Level != level || node.Index != index {
+ return Node{}, false, fmt.Errorf("index of node (%d, %d) points to node (%d, %d)", level, index, node.Level, node.Index)
+ }
+
+ return node, true, nil
}
-// Node returns the node (l, i) if it is built.
-func (t *Tree) Node(level, index int) (Node, bool) {
- node, built := t.nodes[nodeKey{level, index}]
- return node, built
+// Count returns the number of built nodes.
+func (t *Tree) Count() int {
+ return t.order.count
}
-// Nodes returns the built nodes, ordered by level and then by index.
-func (t *Tree) Nodes() []Node {
- nodes := make([]Node, 0, len(t.nodes))
- for _, node := range t.nodes {
- nodes = append(nodes, node)
+// Keys returns the keys of the built nodes, ordered by level and then by index.
+func (t *Tree) Keys() ([]nodeKey, error) {
+ keys := make([]nodeKey, 0, t.order.count)
+ for n := range t.order.count {
+ record, _, err := t.order.Read(n)
+ if err != nil {
+ return nil, err
+ }
+
+ key, _ := decodeTreeRecord(record)
+ keys = append(keys, key)
}
- slices.SortFunc(nodes, func(a, b Node) int {
- return cmp.Or(cmp.Compare(a.Level, b.Level), cmp.Compare(a.Index, b.Index))
+ slices.SortFunc(keys, func(a, b nodeKey) int {
+ return cmp.Or(cmp.Compare(a.level, b.level), cmp.Compare(a.index, b.index))
})
- return nodes
+ return keys, nil
}
// leafSource returns the text that the leaf of message holds or summarizes.
}, true
}
+// decodeTreeRecord reads the key and the position of a record of
+// index/tree.idx.
+func decodeTreeRecord(record []byte) (nodeKey, position) {
+ key := nodeKey{
+ level: int(binary.LittleEndian.Uint32(record[positionBytes+8:])),
+ index: int(binary.LittleEndian.Uint64(record[positionBytes:])),
+ }
+
+ return key, decodePosition(record)
+}
+
// nodeEnd returns the ID after the last message of node (l, i).
func nodeEnd(level, index int) int {
return (index + 1) << level
require.NoError(t, tree.Add(testNode(0, 1, "reply: b")))
require.NoError(t, tree.Add(testNode(1, 0, "a b")))
- node, built := tree.Node(0, 0)
- require.True(t, built)
- require.Equal(t, "user: a", node.Text)
+ require.Equal(t, "user: a", readNode(t, tree, 0, 0).Text)
}
func TestTreeReopenLoadsTheSameNodes(t *testing.T) {
require.NoError(t, tree.Close())
}()
- require.Len(t, tree.nodes, len(nodes))
+ require.Equal(t, len(nodes), tree.Count())
for _, node := range nodes {
- loaded, built := tree.Node(node.Level, node.Index)
- require.True(t, built)
- require.Equal(t, node, loaded)
+ require.Equal(t, node, readNode(t, tree, node.Level, node.Index))
}
+}
+
+func TestTreeBuildsAMissingIndexAgain(t *testing.T) {
+ root := t.TempDir()
+ clock := newTestClock()
+ tree, err := openTree(t.Context(), root, 2, clock.Now)
+ require.NoError(t, err)
+
+ nodes := []Node{
+ testNode(0, 1, "reply: b"),
+ testNode(0, 0, "user: a"),
+ testNode(1, 0, "user: a\nreply: b"),
+ }
+ for _, node := range nodes {
+ require.NoError(t, tree.Add(node))
+ }
+
+ require.NoError(t, tree.Close())
+ require.NoError(t, os.RemoveAll(filepath.Join(root, "index")))
_, err = openTree(t.Context(), root, 1, clock.Now)
require.ErrorContains(t, err, "covers messages after the log")
+ require.NoError(t, os.RemoveAll(filepath.Join(root, "index")))
+
+ tree, err = openTree(t.Context(), root, 2, clock.Now)
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, tree.Close())
+ }()
+
+ require.Equal(t, len(nodes), tree.Count())
+ for _, node := range nodes {
+ require.Equal(t, node, readNode(t, tree, node.Level, node.Index))
+ }
+}
+
+func TestTreeIndexesTheNodesAfterACrash(t *testing.T) {
+ root := t.TempDir()
+ clock := newTestClock()
+ tree, err := openTree(t.Context(), root, 2, clock.Now)
+ require.NoError(t, err)
+
+ require.NoError(t, tree.Add(testNode(0, 0, "user: a")))
+ require.NoError(t, tree.Add(testNode(0, 1, "reply: b")))
+ require.NoError(t, tree.Close())
+
+ // A crash after the line write and in the middle of the index writes.
+ order := filepath.Join(root, "index", "tree.idx")
+ require.NoError(t, os.Truncate(order, treeRecordBytes+5))
+ require.NoError(t, os.Truncate(filepath.Join(root, "index", "tree-0.idx"), positionBytes))
+
+ tree, err = openTree(t.Context(), root, 2, clock.Now)
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, tree.Close())
+ }()
+
+ require.Equal(t, 2, tree.Count())
+ require.Equal(t, "reply: b", readNode(t, tree, 0, 1).Text)
+ require.NoError(t, tree.Add(testNode(1, 0, "user: a\nreply: b")))
+}
+
+// isBuilt tells if the node (l, i) of tree is built.
+func isBuilt(t *testing.T, tree *Tree, level, index int) bool {
+ t.Helper()
+
+ built, err := tree.Built(level, index)
+ require.NoError(t, err)
+
+ return built
+}
+
+// readNode reads the node (l, i) of tree, which must be built.
+func readNode(t *testing.T, tree *Tree, level, index int) Node {
+ t.Helper()
+
+ node, built, err := tree.Node(level, index)
+ require.NoError(t, err)
+ require.True(t, built, "node (%d, %d)", level, index)
+
+ return node
}
func testNode(level, index int, text string) Node {
// nodeReader gives the built nodes of a tree.
type nodeReader interface {
- Node(level, index int) (Node, bool)
+ Built(level, index int) (bool, error)
+ Node(level, index int) (Node, bool, error)
}
// View is a list of tree nodes that covers messages 0 to end-1 without a gap,
// the size is at most low. A batch merges only pairs whose parent is built. If
// the batch cannot get to low, it continues at each new message until it does.
//
-// The size of a line is the size of its node text, or the size of the pending
-// text when the node is not built.
+// The view keeps the nodes of its built lines in memory, so its size and its
+// text need no read of the tree. The size of a line is the size of its node
+// text, or the size of the pending text when the node is not built.
//
// For the sibling pair (l, i) and (l, i+1), due is (T - last) / 2^l, with T the
// number of messages and last the ID of the last message of the pair. That is
high int
low int
lines []nodeKey
+ nodes map[nodeKey]Node
end int
merging bool
}
// newView returns an empty view with the limits high and low.
func newView(high, low int) *View {
return &View{
- high: high,
- low: low,
+ high: high,
+ low: low,
+ nodes: make(map[nodeKey]Node),
}
}
return nil, fmt.Errorf("view line (%d, %d) does not start at message %d", line.level, line.index, v.end)
}
- _, built := tree.Node(line.level, line.index)
+ built, err := v.load(tree, line)
+ if err != nil {
+ return nil, err
+ }
+
if line.level > 0 && !built {
return nil, fmt.Errorf("view line (%d, %d) is not built", line.level, line.index)
}
return v, nil
}
+// load keeps the node of line when it is built, and tells if it is.
+func (v *View) load(tree nodeReader, line nodeKey) (bool, error) {
+ node, built, err := tree.Node(line.level, line.index)
+ if err != nil || !built {
+ return false, err
+ }
+
+ v.nodes[line] = node
+
+ return true, nil
+}
+
// State returns the saved form of the view.
func (v *View) State() viewState {
state := viewState{
// Append adds the leaf of the next message as a line and merges when the view
// is too large. It tells if a merge changed the view.
-func (v *View) Append(tree nodeReader) bool {
- v.lines = append(v.lines, nodeKey{0, v.end})
+func (v *View) Append(tree nodeReader) (bool, error) {
+ line := nodeKey{0, v.end}
+ _, err := v.load(tree, line)
+ if err != nil {
+ return false, err
+ }
+
+ v.lines = append(v.lines, line)
v.end++
return v.fit(tree)
// fit merges the most due pairs while a batch runs, and tells if a merge
// changed the view.
-func (v *View) fit(tree nodeReader) bool {
- size := v.Size(tree)
+func (v *View) fit(tree nodeReader) (bool, error) {
+ size := v.Size()
if size > v.high {
v.merging = true
}
merged := false
for v.merging && size > v.low {
- best := v.mostDue(tree)
- if best < 0 {
- return merged
+ best, err := v.mostDue(tree)
+ if err != nil || best < 0 {
+ return merged, err
}
left, right := v.lines[best], v.lines[best+1]
parent := nodeKey{left.level + 1, left.index / 2}
- size += lineSize(tree, parent) - lineSize(tree, left) - lineSize(tree, right)
+ _, err = v.load(tree, parent)
+ if err != nil {
+ return merged, err
+ }
+
+ size += v.lineSize(parent) - v.lineSize(left) - v.lineSize(right)
+ delete(v.nodes, left)
+ delete(v.nodes, right)
v.lines[best] = parent
v.lines = append(v.lines[:best+1], v.lines[best+2:]...)
merged = true
v.merging = false
- return merged
+ return merged, nil
}
// Coarsen takes each merge of other, a view of the same messages, and then
// the coarser of the two lines that start at its first message. Dyadic ranges
// either nest or do not overlap, so the result covers the messages without a
// gap.
-func (v *View) Coarsen(other *View, tree nodeReader) {
+func (v *View) Coarsen(other *View, tree nodeReader) error {
lines := make([]nodeKey, 0, len(v.lines))
+ nodes := make(map[nodeKey]Node, len(v.nodes))
mine, theirs := 0, 0
for start := 0; start < v.end; {
line := v.lines[mine]
+ node, built := v.nodes[line]
if other.lines[theirs].level > line.level {
line = other.lines[theirs]
+ node, built = other.nodes[line]
}
lines = append(lines, line)
+ if built {
+ nodes[line] = node
+ }
+
start = nodeEnd(line.level, line.index)
for mine < len(v.lines) && v.lines[mine].index<<v.lines[mine].level < start {
mine++
}
v.lines = lines
+ v.nodes = nodes
v.merging = true
- v.fit(tree)
+ _, err := v.fit(tree)
+
+ return err
}
// mostDue returns the position of the left line of the most due sibling pair
// whose parent is built, or -1 when there is no such pair.
-func (v *View) mostDue(tree nodeReader) int {
+func (v *View) mostDue(tree nodeReader) (int, error) {
best := -1
bestDue := 0.0
for n := 0; n+1 < len(v.lines); n++ {
continue
}
- _, built := tree.Node(left.level+1, left.index/2)
- if !built {
+ last := nodeEnd(right.level, right.index) - 1
+ due := math.Ldexp(float64(v.end-last), -left.level)
+ if best >= 0 && due <= bestDue {
continue
}
- last := nodeEnd(right.level, right.index) - 1
- due := math.Ldexp(float64(v.end-last), -left.level)
- if best < 0 || due > bestDue {
+ built, err := tree.Built(left.level+1, left.index/2)
+ if err != nil {
+ return -1, err
+ }
+
+ if built {
best = n
bestDue = due
}
}
- return best
+ return best, nil
+}
+
+// Refresh keeps node when it is the node of a pending line.
+func (v *View) Refresh(node Node) {
+ key := nodeKey{node.Level, node.Index}
+ for n := len(v.lines) - 1; n >= 0; n-- {
+ line := v.lines[n]
+ if line == key {
+ v.nodes[key] = node
+ return
+ }
+
+ if line.index<<line.level <= key.index<<key.level {
+ return
+ }
+ }
}
// Size returns the sum of the line sizes.
-func (v *View) Size(tree nodeReader) int {
+func (v *View) Size() int {
size := 0
for _, line := range v.lines {
- size += lineSize(tree, line)
+ size += v.lineSize(line)
}
return size
}
+// lineSize returns the size of a view line.
+func (v *View) lineSize(line nodeKey) int {
+ node, built := v.nodes[line]
+ if !built {
+ return len(pending)
+ }
+
+ return node.Size
+}
+
// Render returns the view as the model reads it: one line "id+n|text" for each
// node inside <chat> tags, with the newlines of the text as spaces.
-func (v *View) Render(tree nodeReader) string {
+func (v *View) Render() string {
var text strings.Builder
text.WriteString("<chat>\n")
for _, line := range v.lines {
- renderLine(&text, tree, line)
+ v.renderLine(&text, line)
}
text.WriteString("</chat>")
return text.String()
}
+// renderLine writes one view line, "id+n|text", and a newline.
+func (v *View) renderLine(text *strings.Builder, line nodeKey) {
+ body := pending
+ node, built := v.nodes[line]
+ if built {
+ body = strings.ReplaceAll(node.Text, "\n", " ")
+ }
+
+ text.WriteString(strconv.Itoa(line.index << line.level))
+ text.WriteByte('+')
+ text.WriteString(strconv.Itoa(1 << line.level))
+ text.WriteByte('|')
+ text.WriteString(body)
+ text.WriteByte('\n')
+}
+
// Context returns the lines of the view before message end as the model reads
// them. It stops at the first line that is not built, so a compaction never
// sees a pending mark or a part of a message.
-func (v *View) Context(tree nodeReader, end int) string {
+func (v *View) Context(end int) string {
var text strings.Builder
text.WriteString("<chat>\n")
for _, line := range v.lines {
- _, built := tree.Node(line.level, line.index)
+ _, built := v.nodes[line]
if !built || nodeEnd(line.level, line.index) > end {
break
}
- renderLine(&text, tree, line)
+ v.renderLine(&text, line)
}
text.WriteString("</chat>")
return text.String()
}
-
-// lineSize returns the size of a view line.
-func lineSize(tree nodeReader, line nodeKey) int {
- node, built := tree.Node(line.level, line.index)
- if !built {
- return len(pending)
- }
-
- return node.Size
-}
-
-// renderLine writes one view line, "id+n|text", and a newline.
-func renderLine(text *strings.Builder, tree nodeReader, line nodeKey) {
- body := pending
- node, built := tree.Node(line.level, line.index)
- if built {
- body = strings.ReplaceAll(node.Text, "\n", " ")
- }
-
- text.WriteString(strconv.Itoa(line.index << line.level))
- text.WriteByte('+')
- text.WriteString(strconv.Itoa(1 << line.level))
- text.WriteByte('|')
- text.WriteString(body)
- text.WriteByte('\n')
-}
starts := states.starts()
view.high = len(starts)
view.low = len(starts)
- view.Append(tree)
+ _, err := view.Append(tree)
+ require.NoError(t, err)
lines := make([]int, len(view.lines))
for n, line := range view.lines {
view := newView(10, 5)
sizes := make([]int, 0, 20)
for range 20 {
- view.Append(tree)
- sizes = append(sizes, view.Size(tree))
+ _, err := view.Append(tree)
+ require.NoError(t, err)
+ sizes = append(sizes, view.Size())
}
require.Equal(t, []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 5, 6, 7, 8, 9, 10, 5, 6, 7, 8}, sizes)
view := newView(3, 2)
for range 4 {
- view.Append(tree)
+ _, err := view.Append(tree)
+ require.NoError(t, err)
}
require.Equal(t, []nodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}}, view.lines)
require.True(t, view.merging)
tree.sizes[nodeKey{1, 0}] = 1
- view.Append(tree)
+ _, err := view.Append(tree)
+ require.NoError(t, err)
require.Equal(t, []nodeKey{{1, 0}, {0, 2}, {0, 3}, {0, 4}}, view.lines)
require.True(t, view.merging)
tree.sizes[nodeKey{1, 1}] = 1
tree.sizes[nodeKey{1, 2}] = 1
tree.sizes[nodeKey{2, 0}] = 1
- view.Append(tree)
+ _, err = view.Append(tree)
+ require.NoError(t, err)
require.Equal(t, []nodeKey{{2, 0}, {1, 2}}, view.lines)
require.False(t, view.merging)
}
Lines: [][2]int{{1, 0}, {0, 2}},
}, chatHigh, chatLow, tree)
require.NoError(t, err)
- require.Equal(t, "<chat>\n0+2|aaa aaa\n2+1|"+pending+"\n</chat>", view.Render(tree))
- require.Equal(t, 7+len(pending), view.Size(tree))
+ require.Equal(t, "<chat>\n0+2|aaa aaa\n2+1|"+pending+"\n</chat>", view.Render())
+ require.Equal(t, 7+len(pending), view.Size())
}
func TestViewCoarsenTakesTheCoarserLines(t *testing.T) {
}, 100, 100, builtTree{})
require.NoError(t, err)
- view.Coarsen(other, builtTree{})
+ require.NoError(t, view.Coarsen(other, builtTree{}))
require.Equal(t, []nodeKey{{1, 0}, {1, 1}, {0, 4}, {0, 5}}, view.lines)
view.low = 2
- view.Coarsen(other, builtTree{})
+ require.NoError(t, view.Coarsen(other, builtTree{}))
require.Equal(t, []nodeKey{{2, 0}, {1, 2}}, view.lines)
require.False(t, view.merging)
}
Lines: [][2]int{{1, 0}, {0, 2}, {0, 3}},
}, 100, 100, tree)
require.NoError(t, err)
- require.Equal(t, "<chat>\n</chat>", view.Context(tree, 0))
- require.Equal(t, "<chat>\n0+2|a\n</chat>", view.Context(tree, 2))
- require.Equal(t, "<chat>\n0+2|a\n</chat>", view.Context(tree, 3))
+ require.Equal(t, "<chat>\n</chat>", view.Context(0))
+ require.Equal(t, "<chat>\n0+2|a\n</chat>", view.Context(2))
+ require.Equal(t, "<chat>\n0+2|a\n</chat>", view.Context(3))
}
func TestRestoreViewRejectsAGap(t *testing.T) {
// builtTree is a tree with each node built, with 1 byte of text.
type builtTree struct{}
-func (builtTree) Node(level, index int) (Node, bool) {
- return testNode(level, index, "a"), true
+func (builtTree) Built(level, index int) (bool, error) {
+ return true, nil
+}
+
+func (builtTree) Node(level, index int) (Node, bool, error) {
+ return testNode(level, index, "a"), true, nil
}
// partTree is a tree with the built nodes of its sizes.
sizes map[nodeKey]int
}
-func (p partTree) Node(level, index int) (Node, bool) {
+func (p partTree) Built(level, index int) (bool, error) {
+ _, built := p.sizes[nodeKey{level, index}]
+ return built, nil
+}
+
+func (p partTree) Node(level, index int) (Node, bool, error) {
size, built := p.sizes[nodeKey{level, index}]
if !built {
- return Node{}, false
+ return Node{}, false, nil
}
text := strings.Repeat("a", size)
text = "aaa\naaa"
}
- return testNode(level, index, text), true
+ return testNode(level, index, text), true, nil
}
projects / onechat / commitdiff