package ingest
import (
"path/filepath"
"strings"
"testing"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/events"
)
// A repository re-added under the same name finds no publisher cursor,
// publish record, fold mark or durable consumer of the deleted one: a
// surviving cursor over an emptied index would skip the census
// (quarry#294). A neighbour whose name continues the deleted one's —
// foo-bar, or foo- and what reads as a node hash — keeps all of its own.
func TestForgetTakesOnlyThatRepository(t *testing.T) {
ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
kv, err := ev.Cursors()
if err != nil {
t.Fatal(err)
}
// consumer is the durable repo's pipeline on node reads with.
consumer := func(repo, node string) string {
name := durableName(repo, node)
if _, err := ev.JS().AddConsumer(events.StreamName, &nats.ConsumerConfig{
Durable: name, FilterSubject: "zit.evt." + repo + ".>", AckPolicy: nats.AckExplicitPolicy,
}); err != nil {
t.Fatal(err)
}
return name
}
// lookalike's single-node durable is the one foo would have on n3.
lookalike := strings.TrimPrefix(durableName("foo", "n3"), "indexer-")
goneConsumer := consumer("foo", "n2")
keptConsumers := []string{
consumer("foo-bar", ""),
consumer("foo-bar", "n2"),
consumer(lookalike, ""),
consumer(lookalike, "n2"),
}
gone := []string{"foo", "pub.foo", "seen." + durableName("foo", ""),
"seen." + durableName("foo", "n1"), "seen." + goneConsumer}
kept := []string{"foo-bar", "pub.foo-bar", lookalike, "pub." + lookalike, "foo.x", "pub.foo.x"}
for _, c := range keptConsumers {
kept = append(kept, "seen."+c)
}
for _, k := range append(gone, kept...) {
if _, err := kv.Put(k, []byte("{}")); err != nil {
t.Fatal(err)
}
}
// Its store events go too, which a new consumer would otherwise
// fold into the next foo's index; foo.x's events are a different
// repository's however much of foo's subject they share.
keptEvents := []string{"foo-bar", lookalike, "foo.x", "foo.artifact"}
for _, repo := range append([]string{"foo"}, keptEvents...) {
for _, kind := range []string{"artifact", "reset"} {
if _, err := ev.JS().Publish((&Pipeline{code: repo}).subject(kind), []byte("{}")); err != nil {
t.Fatal(err)
}
}
}
if err := Forget(ev, "foo", "n1"); err != nil {
t.Fatal(err)
}
for _, k := range gone {
if _, err := kv.Get(k); err == nil {
t.Errorf("%s survived", k)
}
}
for _, k := range kept {
if _, err := kv.Get(k); err != nil {
t.Errorf("%s went with foo: %v", k, err)
}
}
if _, err := ev.JS().ConsumerInfo(events.StreamName, goneConsumer); err == nil {
t.Errorf("consumer %s survived", goneConsumer)
}
for _, c := range keptConsumers {
if _, err := ev.JS().ConsumerInfo(events.StreamName, c); err != nil {
t.Errorf("consumer %s went with foo: %v", c, err)
}
}
held := func(filter string) uint64 {
info, err := ev.JS().StreamInfo(events.StreamName, &nats.StreamInfoRequest{SubjectsFilter: filter})
if err != nil {
t.Fatal(err)
}
var n uint64
for _, c := range info.State.Subjects {
n += c
}
return n
}
if n := held("zit.evt.foo.*"); n != 0 {
t.Errorf("%d of foo's store events survived", n)
}
for _, repo := range keptEvents {
if n := held((&Pipeline{code: repo}).subject(">")); n != 2 {
t.Errorf("%s holds %d store events after foo's went, want 2", repo, n)
}
}
}