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