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

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