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         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 := newSyncerGatedSync()
  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 := newSyncerGatedSync()
  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(), discardLog())
  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 := syncerRunKilledWriter(t, root, 200+100*round)
 102
 103                 memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
 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, syncerWriterKind(id), message.Kind)
 110                         require.Equal(t, syncerWriterText(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 syncerChildDirVariable
 119 // and prints the ID of each message after its Append returns, until it is
 120 // killed.
 121 func TestWriterChild(t *testing.T) {
 122         root := os.Getenv(syncerChildDirVariable)
 123         if root == "" {
 124                 t.Skip("runs only as a child process")
 125         }
 126
 127         memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
 128         require.NoError(t, err)
 129
 130         for id := memory.log.Len(); ; id++ {
 131                 _, err = memory.Append(syncerWriterKind(id), syncerWriterText(id))
 132                 require.NoError(t, err)
 133
 134                 _, err = fmt.Println(id)
 135                 require.NoError(t, err)
 136         }
 137 }
 138
 139 // SyncerGatedSync is a sync function that blocks until the test releases it.
 140 type SyncerGatedSync struct {
 141         mutex   sync.Mutex
 142         count   int
 143         started chan struct{}
 144         release chan struct{}
 145 }
 146
 147 func (g *SyncerGatedSync) Sync() error {
 148         g.mutex.Lock()
 149         g.count++
 150         g.mutex.Unlock()
 151
 152         g.started <- struct{}{}
 153         <-g.release
 154
 155         return nil
 156 }
 157
 158 // WaitStarted waits until sync number n has started.
 159 func (g *SyncerGatedSync) WaitStarted(t *testing.T, n int) {
 160         t.Helper()
 161
 162         require.Eventually(t, func() bool {
 163                 return g.Count() >= n
 164         }, time.Second, time.Millisecond)
 165 }
 166
 167 // Release lets the running sync end.
 168 func (g *SyncerGatedSync) Release() {
 169         <-g.started
 170         g.release <- struct{}{}
 171 }
 172
 173 func (g *SyncerGatedSync) Count() int {
 174         g.mutex.Lock()
 175         defer g.mutex.Unlock()
 176
 177         return g.count
 178 }
 179
 180 func newSyncerGatedSync() *SyncerGatedSync {
 181         return &SyncerGatedSync{
 182                 started: make(chan struct{}, 100),
 183                 release: make(chan struct{}),
 184         }
 185 }
 186
 187 // syncerRunKilledWriter starts the writer child on the chat in root, kills it
 188 // with SIGKILL after it acknowledged count messages, and returns the number of
 189 // messages that it acknowledged.
 190 func syncerRunKilledWriter(t *testing.T, root string, count int) int {
 191         t.Helper()
 192
 193         command := exec.Command(os.Args[0], "-test.run=^TestWriterChild$")
 194         command.Env = append(os.Environ(), syncerChildDirVariable+"="+root)
 195         stdout, err := command.StdoutPipe()
 196         require.NoError(t, err)
 197         require.NoError(t, command.Start())
 198
 199         acknowledged := 0
 200         scanner := bufio.NewScanner(stdout)
 201         for acknowledged < count && scanner.Scan() {
 202                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 203                 if err == nil {
 204                         acknowledged++
 205                 }
 206         }
 207
 208         require.NoError(t, command.Process.Kill())
 209         for scanner.Scan() {
 210                 _, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
 211                 if err == nil {
 212                         acknowledged++
 213                 }
 214         }
 215
 216         require.Error(t, command.Wait())
 217         require.Equal(t, count, min(acknowledged, count), "the child stopped early")
 218
 219         return acknowledged
 220 }
 221
 222 // syncerWriterKind returns the kind of message id of the writer child.
 223 func syncerWriterKind(id int) string {
 224         return []string{"user", "tool", "echo", "reply"}[id%4]
 225 }
 226
 227 // syncerWriterText returns the text of message id of the writer child. Some
 228 // texts are large, so a kill can come during the write of one line.
 229 func syncerWriterText(id int) string {
 230         size := 40 + id%7*30
 231         if id%5 == 4 {
 232                 size = 64*1024 + id
 233         }
 234
 235         return strings.Repeat(strconv.Itoa(id%10), size)
 236 }