git.giacomolli.uk - onechat/blob - syncer_test.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "errors"
   6         "fmt"
   7         "os"
   8         "os/exec"
   9         "strconv"
  10         "strings"
  11         "sync"
  12         "testing"
  13         "time"
  14
  15         "github.com/stretchr/testify/require"
  16 )
  17
  18 // childDirVariable names the chat directory of the writer child process of
  19 // TestKilledWriterLosesNoLoggedMessage.
  20 const childDirVariable = "ONECHAT_TEST_CHILD_DIR"
  21
  22 func TestFlushWaitsForASyncThatStartsAfterTheWrite(t *testing.T) {
  23         gate := newGatedSync()
  24         s := newSyncer(gate.Sync)
  25         s.Written()
  26         gate.WaitStarted(t, 1)
  27
  28         // This write comes during the first sync, so that sync does not cover
  29         // it.
  30         s.Written()
  31         flushed := make(chan error)
  32         go func() {
  33                 flushed <- s.Flush(t.Context())
  34         }()
  35
  36         gate.Release()
  37         gate.WaitStarted(t, 2)
  38         select {
  39         case err := <-flushed:
  40                 t.Fatalf("flush returned before the second sync: %v", err)
  41         case <-time.After(20 * time.Millisecond):
  42         }
  43
  44         gate.Release()
  45         require.NoError(t, <-flushed)
  46         require.NoError(t, s.Close())
  47         require.Equal(t, 2, gate.Count())
  48 }
  49
  50 func TestOneSyncCoversTheWritesDuringTheSyncBeforeIt(t *testing.T) {
  51         gate := newGatedSync()
  52         s := newSyncer(gate.Sync)
  53         s.Written()
  54         gate.WaitStarted(t, 1)
  55         for range 10 {
  56                 s.Written()
  57         }
  58
  59         gate.Release()
  60         gate.WaitStarted(t, 2)
  61         gate.Release()
  62         require.NoError(t, s.Flush(t.Context()))
  63         require.Equal(t, 2, gate.Count())
  64         require.NoError(t, s.Close())
  65 }
  66
  67 func TestFlushWithoutWritesNeedsNoSync(t *testing.T) {
  68         gate := newGatedSync()
  69         s := newSyncer(gate.Sync)
  70         require.NoError(t, s.Flush(t.Context()))
  71         require.NoError(t, s.Close())
  72         require.Equal(t, 0, gate.Count())
  73 }
  74
  75 func TestAFailedSyncFailsEachLaterFlush(t *testing.T) {
  76         s := newSyncer(func() error {
  77                 return errors.New("disk error")
  78         })
  79
  80         s.Written()
  81         require.ErrorContains(t, s.Flush(t.Context()), "disk error")
  82
  83         s.Written()
  84         require.ErrorContains(t, s.Flush(t.Context()), "disk error")
  85         require.ErrorContains(t, s.Close(), "disk error")
  86 }
  87
  88 func TestMemoryFlushSyncsTheChat(t *testing.T) {
  89         memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer())
  90         require.NoError(t, err)
  91
  92         _, err = memory.Append("tool", "shell: ls")
  93         require.NoError(t, err)
  94         require.NoError(t, memory.Flush(t.Context()))
  95         require.NoError(t, memory.Close())
  96 }
  97
  98 func TestKilledWriterLosesNoLoggedMessage(t *testing.T) {
  99         root := t.TempDir()
 100         for round := range 3 {
 101                 acknowledged := runKilledWriter(t, root, 200+100*round)
 102
 103                 memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer())
 104                 require.NoError(t, err)
 105                 require.GreaterOrEqual(t, memory.log.Len(), acknowledged)
 106                 for id := range memory.log.Len() {
 107                         message, err := memory.log.Message(id)
 108                         require.NoError(t, err)
 109                         require.Equal(t, writerKind(id), message.Kind)
 110                         require.Equal(t, writerText(id), message.Text)
 111                 }
 112
 113                 require.NoError(t, memory.Close())
 114         }
 115 }
 116
 117 // TestWriterChild is the child process of TestKilledWriterLosesNoLoggedMessage.
 118 // It appends messages to the chat in the directory of childDirVariable and
 119 // prints the ID of each message after its Append returns, until it is killed.
 120 func TestWriterChild(t *testing.T) {
 121         root := os.Getenv(childDirVariable)
 122         if root == "" {
 123                 t.Skip("runs only as a child process")
 124         }
 125
 126         memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer())
 127         require.NoError(t, err)
 128
 129         for id := memory.log.Len(); ; id++ {
 130                 _, err = memory.Append(writerKind(id), writerText(id))
 131                 require.NoError(t, err)
 132
 133                 _, err = fmt.Println(id)
 134                 require.NoError(t, err)
 135         }
 136 }
 137
 138 // gatedSync is a sync function that blocks until the test releases it.
 139 type gatedSync struct {
 140         mutex   sync.Mutex
 141         count   int
 142         started chan struct{}
 143         release chan struct{}
 144 }
 145
 146 func newGatedSync() *gatedSync {
 147         return &gatedSync{
 148                 started: make(chan struct{}, 100),
 149                 release: make(chan struct{}),
 150         }
 151 }
 152
 153 func (g *gatedSync) Sync() error {
 154         g.mutex.Lock()
 155         g.count++
 156         g.mutex.Unlock()
 157
 158         g.started <- struct{}{}
 159         <-g.release
 160
 161         return nil
 162 }
 163
 164 // WaitStarted waits until sync number n has started.
 165 func (g *gatedSync) WaitStarted(t *testing.T, n int) {
 166         t.Helper()
 167
 168         require.Eventually(t, func() bool {
 169                 return g.Count() >= n
 170         }, time.Second, time.Millisecond)
 171 }
 172
 173 // Release lets the running sync end.
 174 func (g *gatedSync) Release() {
 175         <-g.started
 176         g.release <- struct{}{}
 177 }
 178
 179 func (g *gatedSync) Count() int {
 180         g.mutex.Lock()
 181         defer g.mutex.Unlock()
 182
 183         return g.count
 184 }
 185
 186 // runKilledWriter starts the writer child on the chat in root, kills it with
 187 // SIGKILL after it acknowledged count messages, and returns the number of
 188 // messages that it acknowledged.
 189 func runKilledWriter(t *testing.T, root string, count int) int {
 190         t.Helper()
 191
 192         command := exec.Command(os.Args[0], "-test.run=^TestWriterChild$")
 193         command.Env = append(os.Environ(), childDirVariable+"="+root)
 194         stdout, err := command.StdoutPipe()
 195         require.NoError(t, err)
 196         require.NoError(t, command.Start())
 197
 198         acknowledged := 0
 199         scanner := bufio.NewScanner(stdout)
 200         for acknowledged < count && scanner.Scan() {
 201                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 202                 if err == nil {
 203                         acknowledged++
 204                 }
 205         }
 206
 207         require.NoError(t, command.Process.Kill())
 208         for scanner.Scan() {
 209                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 210                 if err == nil {
 211                         acknowledged++
 212                 }
 213         }
 214
 215         require.Error(t, command.Wait())
 216         require.Equal(t, count, min(acknowledged, count), "the child stopped early")
 217
 218         return acknowledged
 219 }
 220
 221 // writerKind returns the kind of message id of the writer child.
 222 func writerKind(id int) string {
 223         return []string{"user", "tool", "echo", "reply"}[id%4]
 224 }
 225
 226 // writerText returns the text of message id of the writer child. Some texts
 227 // are large, so a kill can come during the write of one line.
 228 func writerText(id int) string {
 229         size := 40 + id%7*30
 230         if id%5 == 4 {
 231                 size = 64*1024 + id
 232         }
 233
 234         return strings.Repeat(strconv.Itoa(id%10), size)
 235 }