83 lines
2.9 KiB
Go
83 lines
2.9 KiB
Go
package busdedup
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// **The same announcement published twice with one message id is stored once** (novox/hq ADR 0406 §6, issue 976). The
|
|
// forge module announces a merge, a head and a close with a deterministic `x-event-id` (announce.ts); the runtime makes
|
|
// that header the bus's message id (mesh-tools node-tools/internal/bus/bus.go, `nats.MsgId`), and the events stream's
|
|
// message de-duplication keeps the second. The publish here is the runtime's own: the headers the module stamps, the id
|
|
// as the message id. A real JetStream server is linked into the test and started for it alone (the TypeScript harness
|
|
// has none, so this is the Go half of the module's tests).
|
|
|
|
const subject = "mesh.mod.gitea.event.pull.merged"
|
|
|
|
func publish(t *testing.T, js nats.JetStreamContext, id string) *nats.PubAck {
|
|
t.Helper()
|
|
msg := nats.NewMsg(subject)
|
|
msg.Header.Set("x-event-id", id)
|
|
msg.Header.Set("x-source", "gitea")
|
|
msg.Header.Set("content-type", "application/json")
|
|
msg.Data = []byte(`{"owner":"novox","repo":"app","number":7,"merge_commit_sha":"mmm222"}`)
|
|
ack, err := js.PublishMsg(msg, nats.MsgId(id))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return ack
|
|
}
|
|
|
|
func TestTheSameAnnouncementPublishedTwiceWithOneMessageIdIsStoredOnce(t *testing.T) {
|
|
s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true,
|
|
StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
go s.Start()
|
|
if !s.ReadyForConnections(30 * time.Second) {
|
|
t.Fatal("a bus for this test did not come up within 30s")
|
|
}
|
|
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
|
|
nc, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, err := nc.JetStream()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The events stream as the controller makes it: two minutes of message de-duplication.
|
|
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.>"}, Duplicates: 2 * time.Minute}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const id = "merged:novox/app#7@mmm222"
|
|
first := publish(t, js, id)
|
|
second := publish(t, js, id) // the poll and the forge's event both announced it
|
|
if first.Duplicate || !second.Duplicate {
|
|
t.Fatalf("the first publish was a duplicate: %v; the second: %v (wanted false, true)", first.Duplicate, second.Duplicate)
|
|
}
|
|
publish(t, js, "merged:novox/app#8@nnn333") // another merge is another message
|
|
jsNew, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stream, err := jsNew.Stream(context.Background(), "EVENTS")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
info, err := stream.Info(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if info.State.Msgs != 2 {
|
|
t.Fatalf("the stream stores %d messages for two announcements published three times; wanted 2", info.State.Msgs)
|
|
}
|
|
}
|