From: Anderson A. Giacomolli Date: Fri, 6 Nov 2026 21:57:56 +0000 (-0300) Subject: Make the chat writes durable by group commit X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/2532718d5288e1a2586fe5716cc41bb2a0210e16?ds=inline Make the chat writes durable by group commit Each write of a chat still goes to the operating system at once, so a crash of the process loses nothing. After each write, a background syncfs of the file system of the chat starts unless one runs already, and the next one starts when it ends. One sync makes every earlier write durable, so a power loss loses at most the writes after the start of the last finished sync, and there is never more than one sync at a time. Memory.Flush returns when the writes before it are durable, so a caller can make a tool call durable before the tool runs. A failed sync stays and fails each later flush, because the kernel can drop the data of a failed write back. A crash can stop a write in the middle of a line. The open now cuts a part of a line at the end of the last day file, whose write never returned, and still refuses one in an earlier file. golang.org/x/sys gives syncfs on every Linux architecture. The frozen syscall package has no syncfs number on amd64. A test kills a writer child process three times during writes of small and large messages and finds every acknowledged message after each open. Issue: 5daf23e --- diff --git a/go.mod b/go.mod index 2bd2883..673fbdc 100644 --- a/go.mod +++ b/go.mod @@ -2,6 +2,9 @@ module onechat 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 diff --git a/go.sum b/go.sum index 054a196..8de8697 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ 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= diff --git a/journal.go b/journal.go index ef49c6e..cda9375 100644 --- a/journal.go +++ b/journal.go @@ -133,7 +133,9 @@ func (j *journal) Close() error { } // 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 { @@ -141,7 +143,7 @@ func (j *journal) Scan(ctx context.Context, place position, add func([]byte, pos } 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 @@ -156,7 +158,7 @@ func (j *journal) Scan(ctx context.Context, place position, add func([]byte, pos 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 } @@ -165,8 +167,9 @@ func (j *journal) Scan(ctx context.Context, place position, add func([]byte, pos 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) @@ -192,6 +195,10 @@ func scanFile(ctx context.Context, path string, day uint32, offset uint64, add f 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) } @@ -215,6 +222,16 @@ func scanFile(ctx context.Context, path string, day uint32, offset uint64, add f } } +// 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) diff --git a/log_test.go b/log_test.go index 89b87b3..97d28a5 100644 --- a/log_test.go +++ b/log_test.go @@ -141,6 +141,58 @@ func TestLogRejectsAGapInTheIDs(t *testing.T) { 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) diff --git a/memory.go b/memory.go index c3e9f45..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 @@ -457,6 +474,7 @@ func (m *Memory) run(key nodeKey, task Task) { Text: text, Size: len(text), }) + m.syncer.Written() } if err != nil { @@ -490,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{ diff --git a/nix/package.nix b/nix/package.nix index 47cb139..20bb053 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -9,7 +9,7 @@ buildGoModule { inherit src; subPackages = [ "." ]; - vendorHash = "sha256-/h5U9uZRK5I2DkM5xETswrg0ouuLfMUz7LBOWP+edgw="; + vendorHash = "sha256-Mpw1XigL2ARm8oTKwutBLHxdLChSfK1ZbGRaNww8jkk="; doInstallCheck = true; installCheckPhase = '' diff --git a/syncer.go b/syncer.go new file mode 100644 index 0000000..6b69f86 --- /dev/null +++ b/syncer.go @@ -0,0 +1,143 @@ +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 +} diff --git a/syncer_test.go b/syncer_test.go new file mode 100644 index 0000000..28a2597 --- /dev/null +++ b/syncer_test.go @@ -0,0 +1,235 @@ +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) +}