go 1.26.5
-require github.com/stretchr/testify v1.12.0
+require (
+ github.com/stretchr/testify v1.12.0
+ golang.org/x/sys v0.49.0
+)
require gopkg.in/yaml.v3 v3.0.1 // indirect
github.com/stretchr/testify v1.12.0 h1:K6Mr6jO9JICuend/5xzTM03ydSV3vdNRYAdPSukj8uI=
github.com/stretchr/testify v1.12.0/go.mod h1:bOYBZb5qJ00vPzWfIqBUZPaxK8jWiXc6d3ErP4Ca9Gw=
+golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
+golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
}
// 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.
+// position, oldest day first. The zero position starts at the first line. A
+// crash during a write can leave a part of a line at the end of the last day
+// file. Its write never returned, so Scan cuts it off.
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 {
}
slices.Sort(paths)
- for _, path := range paths {
+ for n, path := range paths {
day, err := dayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
if err != nil {
return err
offset = place.offset
}
- err = scanFile(ctx, path, day, offset, add)
+ err = scanFile(ctx, path, day, offset, n == len(paths)-1, add)
if err != nil {
return err
}
return nil
}
-// 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) {
+// scanFile gives each line of the day file at path from offset on to add. In
+// the last day file, it cuts a part of a line at the end.
+func scanFile(ctx context.Context, path string, day uint32, offset uint64, last bool, add func([]byte, position) error) (err error) {
file, err := os.Open(path)
if err != nil {
return fmt.Errorf("open day file: %w", err)
return nil
}
+ if errors.Is(err, io.EOF) && last {
+ return cutFile(path, offset)
+ }
+
if errors.Is(err, io.EOF) {
return fmt.Errorf("%s ends in an incomplete line at %d", filepath.Base(path), offset)
}
}
}
+// cutFile cuts the file at path to size bytes.
+func cutFile(path string, size uint64) error {
+ err := os.Truncate(path, int64(size))
+ if err != nil {
+ return fmt.Errorf("cut incomplete line: %w", err)
+ }
+
+ return nil
+}
+
// 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)
require.ErrorContains(t, err, "expected message 0, found 1")
}
+func TestLogCutsAnIncompleteLastLine(t *testing.T) {
+ root := t.TempDir()
+ clock := newTestClock()
+ log, err := openLog(t.Context(), root, clock.Now)
+ require.NoError(t, err)
+
+ _, err = log.Append("user", "whole")
+ require.NoError(t, err)
+ require.NoError(t, log.Close())
+
+ path := filepath.Join(root, "main", "2026-01-01.jsonl")
+ file, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0)
+ require.NoError(t, err)
+
+ _, err = file.WriteString(`{"i":1,"kind":"user","te`)
+ require.NoError(t, err)
+ require.NoError(t, file.Close())
+
+ log, err = openLog(t.Context(), root, clock.Now)
+ require.NoError(t, err)
+ require.Equal(t, 1, log.Len())
+
+ message, err := log.Append("user", "after the cut")
+ require.NoError(t, err)
+ require.Equal(t, 1, message.ID)
+ require.NoError(t, log.Close())
+
+ log, err = openLog(t.Context(), root, clock.Now)
+ require.NoError(t, err)
+
+ defer func() {
+ require.NoError(t, log.Close())
+ }()
+
+ message, err = log.Message(1)
+ require.NoError(t, err)
+ require.Equal(t, "after the cut", message.Text)
+}
+
+func TestLogRejectsAnIncompleteLineBeforeTheLastDay(t *testing.T) {
+ root := t.TempDir()
+ require.NoError(t, os.MkdirAll(filepath.Join(root, "main"), 0o700))
+
+ first := `{"i":0,"kind":"user","text":"x","size":1,"date":"2026-01-01T12:00:00Z"}` + "\n" + `{"i":1,"kin`
+ second := `{"i":1,"kind":"user","text":"y","size":1,"date":"2026-01-02T12:00:00Z"}` + "\n"
+ require.NoError(t, os.WriteFile(filepath.Join(root, "main", "2026-01-01.jsonl"), []byte(first), 0o600))
+ require.NoError(t, os.WriteFile(filepath.Join(root, "main", "2026-01-02.jsonl"), []byte(second), 0o600))
+
+ _, err := openLog(t.Context(), root, newTestClock().Now)
+ require.ErrorContains(t, err, "ends in an incomplete line")
+}
+
// truncateToOneAndAHalf cuts the index at path in its second record.
func truncateToOneAndAHalf(path string) error {
return os.Truncate(path, positionBytes+positionBytes/2)
// 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
mutex sync.Mutex
root string
log *Log
+ syncer *syncer
tree *Tree
chat *View
cview *View
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())
}
m.mutex.Lock()
+ m.syncer.Written()
m.dispatch()
m.mutex.Unlock()
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()
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())
return Message{}, err
}
+ defer m.syncer.Written()
+
err = m.addMessage(message)
if err != nil {
return Message{}, err
Text: text,
Size: len(text),
})
+ m.syncer.Written()
}
if err != nil {
}
}
+// 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{
inherit src;
subPackages = [ "." ];
- vendorHash = "sha256-/h5U9uZRK5I2DkM5xETswrg0ouuLfMUz7LBOWP+edgw=";
+ vendorHash = "sha256-Mpw1XigL2ARm8oTKwutBLHxdLChSfK1ZbGRaNww8jkk=";
doInstallCheck = true;
installCheckPhase = ''
--- /dev/null
+package main
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "sync"
+
+ "golang.org/x/sys/unix"
+)
+
+// syncer makes the writes of a chat durable by group commit. Each write goes to
+// the operating system at once, so a crash of the process loses nothing. After
+// each write, Written starts a background sync of the file system of the chat,
+// unless one runs already. Then the next sync starts when the running one ends.
+// One sync makes every write before its start durable, so a sync covers many
+// writes, and there is never more than one sync at a time. A power loss loses at
+// most the writes after the start of the last finished sync.
+//
+// A failed sync can mean that the kernel dropped written data, so the error
+// stays, and each later Flush returns it.
+type syncer struct {
+ mutex sync.Mutex
+ sync func() error
+ file *os.File
+ written uint64
+ durable uint64
+ running bool
+ done chan struct{}
+ failure error
+ workers sync.WaitGroup
+}
+
+// openSyncer returns the syncer of the file system that holds dir. It syncs
+// with syncfs, which writes back only that file system.
+func openSyncer(dir string) (*syncer, error) {
+ file, err := os.Open(dir)
+ if err != nil {
+ return nil, fmt.Errorf("open chat directory: %w", err)
+ }
+
+ s := newSyncer(func() error {
+ return unix.Syncfs(int(file.Fd()))
+ })
+ s.file = file
+
+ return s, nil
+}
+
+// newSyncer returns a syncer that makes the writes durable with sync.
+func newSyncer(sync func() error) *syncer {
+ return &syncer{
+ sync: sync,
+ done: make(chan struct{}),
+ }
+}
+
+// Written tells the syncer that a write went to the operating system, and
+// starts a sync when none runs.
+func (s *syncer) Written() {
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ s.written++
+ s.start()
+}
+
+// start starts a sync of the writes so far when none runs and a write is not
+// durable. The caller holds the mutex.
+func (s *syncer) start() {
+ if s.running || s.durable == s.written || s.failure != nil {
+ return
+ }
+
+ s.running = true
+ s.workers.Add(1)
+ go s.run(s.written)
+}
+
+// run syncs once, so the writes up to target become durable, and starts the
+// next sync when writes came during this one.
+func (s *syncer) run(target uint64) {
+ defer s.workers.Done()
+
+ err := s.sync()
+
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+
+ s.running = false
+ if err != nil {
+ s.failure = fmt.Errorf("sync chat files: %w", err)
+ } else {
+ s.durable = target
+ }
+
+ close(s.done)
+ s.done = make(chan struct{})
+ s.start()
+}
+
+// Flush returns when every write before the call is durable.
+func (s *syncer) Flush(ctx context.Context) error {
+ s.mutex.Lock()
+ target := s.written
+ s.mutex.Unlock()
+
+ for {
+ s.mutex.Lock()
+ reached := s.durable >= target
+ failure := s.failure
+ done := s.done
+ s.mutex.Unlock()
+
+ if reached {
+ return nil
+ }
+
+ if failure != nil {
+ return failure
+ }
+
+ select {
+ case <-done:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ }
+}
+
+// Close makes every write durable, waits for the syncs, and closes the
+// directory.
+func (s *syncer) Close() error {
+ err := s.Flush(context.Background())
+ s.workers.Wait()
+ if s.file != nil {
+ err = errors.Join(err, s.file.Close())
+ s.file = nil
+ }
+
+ return err
+}
--- /dev/null
+package main
+
+import (
+ "bufio"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+// childDirVariable names the chat directory of the writer child process of
+// TestKilledWriterLosesNoLoggedMessage.
+const childDirVariable = "ONECHAT_TEST_CHILD_DIR"
+
+func TestFlushWaitsForASyncThatStartsAfterTheWrite(t *testing.T) {
+ gate := newGatedSync()
+ s := newSyncer(gate.Sync)
+ s.Written()
+ gate.WaitStarted(t, 1)
+
+ // This write comes during the first sync, so that sync does not cover
+ // it.
+ s.Written()
+ flushed := make(chan error)
+ go func() {
+ flushed <- s.Flush(t.Context())
+ }()
+
+ gate.Release()
+ gate.WaitStarted(t, 2)
+ select {
+ case err := <-flushed:
+ t.Fatalf("flush returned before the second sync: %v", err)
+ case <-time.After(20 * time.Millisecond):
+ }
+
+ gate.Release()
+ require.NoError(t, <-flushed)
+ require.NoError(t, s.Close())
+ require.Equal(t, 2, gate.Count())
+}
+
+func TestOneSyncCoversTheWritesDuringTheSyncBeforeIt(t *testing.T) {
+ gate := newGatedSync()
+ s := newSyncer(gate.Sync)
+ s.Written()
+ gate.WaitStarted(t, 1)
+ for range 10 {
+ s.Written()
+ }
+
+ gate.Release()
+ gate.WaitStarted(t, 2)
+ gate.Release()
+ require.NoError(t, s.Flush(t.Context()))
+ require.Equal(t, 2, gate.Count())
+ require.NoError(t, s.Close())
+}
+
+func TestFlushWithoutWritesNeedsNoSync(t *testing.T) {
+ gate := newGatedSync()
+ s := newSyncer(gate.Sync)
+ require.NoError(t, s.Flush(t.Context()))
+ require.NoError(t, s.Close())
+ require.Equal(t, 0, gate.Count())
+}
+
+func TestAFailedSyncFailsEachLaterFlush(t *testing.T) {
+ s := newSyncer(func() error {
+ return errors.New("disk error")
+ })
+
+ s.Written()
+ require.ErrorContains(t, s.Flush(t.Context()), "disk error")
+
+ s.Written()
+ require.ErrorContains(t, s.Flush(t.Context()), "disk error")
+ require.ErrorContains(t, s.Close(), "disk error")
+}
+
+func TestMemoryFlushSyncsTheChat(t *testing.T) {
+ memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer())
+ require.NoError(t, err)
+
+ _, err = memory.Append("tool", "shell: ls")
+ require.NoError(t, err)
+ require.NoError(t, memory.Flush(t.Context()))
+ require.NoError(t, memory.Close())
+}
+
+func TestKilledWriterLosesNoLoggedMessage(t *testing.T) {
+ root := t.TempDir()
+ for round := range 3 {
+ acknowledged := runKilledWriter(t, root, 200+100*round)
+
+ memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer())
+ require.NoError(t, err)
+ require.GreaterOrEqual(t, memory.log.Len(), acknowledged)
+ for id := range memory.log.Len() {
+ message, err := memory.log.Message(id)
+ require.NoError(t, err)
+ require.Equal(t, writerKind(id), message.Kind)
+ require.Equal(t, writerText(id), message.Text)
+ }
+
+ require.NoError(t, memory.Close())
+ }
+}
+
+// TestWriterChild is the child process of TestKilledWriterLosesNoLoggedMessage.
+// It appends messages to the chat in the directory of childDirVariable and
+// prints the ID of each message after its Append returns, until it is killed.
+func TestWriterChild(t *testing.T) {
+ root := os.Getenv(childDirVariable)
+ if root == "" {
+ t.Skip("runs only as a child process")
+ }
+
+ memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer())
+ require.NoError(t, err)
+
+ for id := memory.log.Len(); ; id++ {
+ _, err = memory.Append(writerKind(id), writerText(id))
+ require.NoError(t, err)
+
+ _, err = fmt.Println(id)
+ require.NoError(t, err)
+ }
+}
+
+// gatedSync is a sync function that blocks until the test releases it.
+type gatedSync struct {
+ mutex sync.Mutex
+ count int
+ started chan struct{}
+ release chan struct{}
+}
+
+func newGatedSync() *gatedSync {
+ return &gatedSync{
+ started: make(chan struct{}, 100),
+ release: make(chan struct{}),
+ }
+}
+
+func (g *gatedSync) Sync() error {
+ g.mutex.Lock()
+ g.count++
+ g.mutex.Unlock()
+
+ g.started <- struct{}{}
+ <-g.release
+
+ return nil
+}
+
+// WaitStarted waits until sync number n has started.
+func (g *gatedSync) WaitStarted(t *testing.T, n int) {
+ t.Helper()
+
+ require.Eventually(t, func() bool {
+ return g.Count() >= n
+ }, time.Second, time.Millisecond)
+}
+
+// Release lets the running sync end.
+func (g *gatedSync) Release() {
+ <-g.started
+ g.release <- struct{}{}
+}
+
+func (g *gatedSync) Count() int {
+ g.mutex.Lock()
+ defer g.mutex.Unlock()
+
+ return g.count
+}
+
+// runKilledWriter starts the writer child on the chat in root, kills it with
+// SIGKILL after it acknowledged count messages, and returns the number of
+// messages that it acknowledged.
+func runKilledWriter(t *testing.T, root string, count int) int {
+ t.Helper()
+
+ command := exec.Command(os.Args[0], "-test.run=^TestWriterChild$")
+ command.Env = append(os.Environ(), childDirVariable+"="+root)
+ stdout, err := command.StdoutPipe()
+ require.NoError(t, err)
+ require.NoError(t, command.Start())
+
+ acknowledged := 0
+ scanner := bufio.NewScanner(stdout)
+ for acknowledged < count && scanner.Scan() {
+ _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
+ if err == nil {
+ acknowledged++
+ }
+ }
+
+ require.NoError(t, command.Process.Kill())
+ for scanner.Scan() {
+ _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
+ if err == nil {
+ acknowledged++
+ }
+ }
+
+ require.Error(t, command.Wait())
+ require.Equal(t, count, min(acknowledged, count), "the child stopped early")
+
+ return acknowledged
+}
+
+// writerKind returns the kind of message id of the writer child.
+func writerKind(id int) string {
+ return []string{"user", "tool", "echo", "reply"}[id%4]
+}
+
+// writerText returns the text of message id of the writer child. Some texts
+// are large, so a kill can come during the write of one line.
+func writerText(id int) string {
+ size := 40 + id%7*30
+ if id%5 == 4 {
+ size = 64*1024 + id
+ }
+
+ return strings.Repeat(strconv.Itoa(id%10), size)
+}
projects / onechat / commitdiff