package ingest
import (
"errors"
"testing"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/events"
)
// A write that still holds a deleted repository's pipeline must not
// publish, save a cursor or make a consumer once the pipeline is
// stopped: whatever it wrote after the delete's purge, a repository
// created later under the name would inherit (quarry#294). And Stop
// returns only once a round already running has finished, so the
// delete's purge comes after everything that round wrote.
func TestAStoppedPipelineWritesNothing(t *testing.T) {
p, _, repo := testPipeline(t)
for range 2 {
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
}
held := func() uint64 {
info, err := p.ev.JS().StreamInfo(events.StreamName, &nats.StreamInfoRequest{SubjectsFilter: p.subject("*")})
if err != nil {
t.Fatal(err)
}
var n uint64
for _, c := range info.State.Subjects {
n += c
}
return n
}
kv, err := p.ev.Cursors()
if err != nil {
t.Fatal(err)
}
cursor := func() uint64 {
e, err := kv.Get(p.code)
if err != nil {
t.Fatal(err)
}
return e.Revision()
}
before, rev := held(), cursor()
p.Stop()
commitFile(t, repo, "c.txt", "gamma\n", "after the delete")
if _, err := p.Sync(); !errors.Is(err, ErrStopped) {
t.Errorf("a sync after Stop answered %v, want ErrStopped", err)
}
if n := held(); n != before {
t.Errorf("a stopped pipeline published: the stream went from %d to %d", before, n)
}
if r := cursor(); r != rev {
t.Errorf("a stopped pipeline saved its cursor: revision %d to %d", rev, r)
}
}
func TestStopWaitsForARunningRound(t *testing.T) {
p, _, _ := testPipeline(t)
p.mu.Lock() // a round in progress
stopped := make(chan struct{})
go func() {
p.Stop()
close(stopped)
}()
select {
case <-stopped:
t.Fatal("Stop returned while a round was still running")
case <-time.After(200 * time.Millisecond):
}
p.mu.Unlock()
select {
case <-stopped:
case <-time.After(5 * time.Second):
t.Fatal("Stop never returned after the round finished")
}
}