// This write comes during the first sync, so that sync does not cover
// it.
s.Written()
+
flushed := make(chan error)
go func() {
flushed <- s.Flush(t.Context())
gate.Release()
gate.WaitStarted(t, 2)
+
select {
case err := <-flushed:
t.Fatalf("flush returned before the second sync: %v", err)
s := newSyncer(gate.Sync)
s.Written()
gate.WaitStarted(t, 1)
+
for range 10 {
s.Written()
}
memory, err := openMemory(t.Context(), root, time.Now, newFakeSummarizer(), discardLog())
require.NoError(t, err)
require.GreaterOrEqual(t, memory.log.Len(), acknowledged)
+
for id := range memory.log.Len() {
message, err := memory.log.Message(id)
require.NoError(t, err)
func (g *SyncerGatedSync) Sync() error {
g.mutex.Lock()
+
g.count++
g.mutex.Unlock()
g.started <- struct{}{}
+
<-g.release
return nil
// Release lets the running sync end.
func (g *SyncerGatedSync) Release() {
<-g.started
+
g.release <- struct{}{}
}
}
require.NoError(t, command.Process.Kill())
+
for scanner.Scan() {
_, err = strconv.Atoi(strings.TrimSpace(scanner.Text()))
if err == nil {