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         s := newSyncer(func() error {
  80                 return errors.New("disk error")
  81         })
  82
  83         s.Written()
  84         require.ErrorContains(t, s.Flush(t.Context()), "disk error")
  85
  86         s.Written()
  87         require.ErrorContains(t, s.Flush(t.Context()), "disk error")
  88         require.ErrorContains(t, s.Close(), "disk error")
  89 }
  90
  91 func TestMemoryFlushSyncsTheChat(t *testing.T) {
  92         memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog())
  93         require.NoError(t, err)
  94
  95         _, err = memory.Append("tool", "shell: ls")
  96         require.NoError(t, err)
  97         require.NoError(t, memory.Flush(t.Context()))
  98         require.NoError(t, memory.Close())
  99 }
 100
 101 func TestKilledWriterLosesNoLoggedMessage(t *testing.T) {
 102         root := t.TempDir()
 103         for round := range 3 {
 104                 acknowledged := syncerRunKilledWriter(t, root, 200+100*round)
 105
 106                 memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
 107                 require.NoError(t, err)
 108                 require.GreaterOrEqual(t, memory.log.Len(), acknowledged)
 109
 110                 for id := range memory.log.Len() {
 111                         message, err := memory.log.Message(id)
 112                         require.NoError(t, err)
 113                         require.Equal(t, syncerWriterKind(id), message.Kind)
 114                         require.Equal(t, syncerWriterText(id), message.Text)
 115                 }
 116
 117                 require.NoError(t, memory.Close())
 118         }
 119 }
 120
 121 // TestWriterChild is the child process of TestKilledWriterLosesNoLoggedMessage.
 122 // It appends messages to the chat in the directory of syncerChildDirVariable
 123 // and prints the ID of each message after its Append returns, until it is
 124 // killed.
 125 func TestWriterChild(t *testing.T) {
 126         root := os.Getenv(syncerChildDirVariable)
 127         if root == "" {
 128                 t.Skip("runs only as a child process")
 129         }
 130
 131         memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
 132         require.NoError(t, err)
 133
 134         for id := memory.log.Len(); ; id++ {
 135                 _, err = memory.Append(syncerWriterKind(id), syncerWriterText(id))
 136                 require.NoError(t, err)
 137
 138                 _, err = fmt.Println(id)
 139                 require.NoError(t, err)
 140         }
 141 }
 142
 143 // SyncerGatedSync is a sync function that blocks until the test releases it.
 144 type SyncerGatedSync struct {
 145         mutex   sync.Mutex
 146         count   int
 147         started chan struct{}
 148         release chan struct{}
 149 }
 150
 151 func (g *SyncerGatedSync) Sync() error {
 152         g.mutex.Lock()
 153
 154         g.count++
 155         g.mutex.Unlock()
 156
 157         g.started <- struct{}{}
 158
 159         <-g.release
 160
 161         return nil
 162 }
 163
 164 // WaitStarted waits until sync number n has started.
 165 func (g *SyncerGatedSync) 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 *SyncerGatedSync) Release() {
 175         <-g.started
 176
 177         g.release <- struct{}{}
 178 }
 179
 180 func (g *SyncerGatedSync) Count() int {
 181         g.mutex.Lock()
 182         defer g.mutex.Unlock()
 183
 184         return g.count
 185 }
 186
 187 func newSyncerGatedSync() *SyncerGatedSync {
 188         return &SyncerGatedSync{
 189                 started: make(chan struct{}, 100),
 190                 release: make(chan struct{}),
 191         }
 192 }
 193
 194 // syncerRunKilledWriter starts the writer child on the chat in root, kills it
 195 // with SIGKILL after it acknowledged count messages, and returns the number of
 196 // messages that it acknowledged.
 197 func syncerRunKilledWriter(t *testing.T, root string, count int) int {
 198         t.Helper()
 199
 200         command := exec.Command(os.Args[0], "-test.run=^TestWriterChild$")
 201         command.Env = append(os.Environ(), syncerChildDirVariable+"="+root)
 202         stdout, err := command.StdoutPipe()
 203         require.NoError(t, err)
 204         require.NoError(t, command.Start())
 205
 206         acknowledged := 0
 207         scanner := bufio.NewScanner(stdout)
 208         for acknowledged < count && scanner.Scan() {
 209                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 210                 if err == nil {
 211                         acknowledged++
 212                 }
 213         }
 214
 215         require.NoError(t, command.Process.Kill())
 216
 217         for scanner.Scan() {
 218                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 219                 if err == nil {
 220                         acknowledged++
 221                 }
 222         }
 223
 224         require.Error(t, command.Wait())
 225         require.Equal(t, count, min(acknowledged, count), "the child stopped early")
 226
 227         return acknowledged
 228 }
 229
 230 // syncerWriterKind returns the kind of message id of the writer child.
 231 func syncerWriterKind(id int) string {
 232         return []string{"user", "tool", "echo", "reply"}[id%4]
 233 }
 234
 235 // syncerWriterText returns the text of message id of the writer child. Some
 236 // texts are large, so a kill can come during the write of one line.
 237 func syncerWriterText(id int) string {
 238         size := 40 + id%7*30
 239         if id%5 == 4 {
 240                 size = 64*1024 + id
 241         }
 242
 243         return strings.Repeat(strconv.Itoa(id%10), size)
 244 }