247 lines
8.4 KiB
Go
247 lines
8.4 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
type sigOf struct{}
|
|
|
|
func (sigOf) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil }
|
|
|
|
// 1696/AC7 (ADR 0486 §1): on a real bus with its default largest message of 1 MiB, a declaration whose files add up to more than
|
|
// 3 MiB is sent as references, its objects are in the bucket, each fetched back whole and verified against the digest and size the
|
|
// declaration names; the same declaration inline is refused by the controller, naming size and limit. A changed byte in an object
|
|
// fails the verification.
|
|
func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) { // 1696/AC7
|
|
s := testbus.Start(t)
|
|
conn, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
nc, err := conn.JetStream()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := nc.AddStream(&nats.StreamConfig{Name: "NODES", Subjects: []string{"mesh.node.*.declare"},
|
|
MaxMsgsPerSubject: 1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
defer cancel()
|
|
api, err := jetstream.New(conn)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Run twice: an assertion of an existing bucket is no error and no change.
|
|
if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
|
|
t.Fatalf("asserting the bucket again: %v", err)
|
|
}
|
|
|
|
var resources []map[string]any
|
|
for i := 0; i < 3; i++ {
|
|
resources = append(resources, aFileResource("big-"+string(rune('a'+i)), contentOf("big"+string(rune('a'+i)), 1200<<10)))
|
|
}
|
|
resources = append(resources, aFileResource("small", "tiny\n"))
|
|
inline := sendable{Resources: resources}
|
|
byDigest := sendable{fileReferences: true, Resources: resources}
|
|
|
|
bus := link.OverNATS{Conn: conn, JS: nc}
|
|
body, err := inline.Body()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := link.Declare(ctx, bus, sigOf{}, "novox", body, 5*time.Second); err == nil ||
|
|
!strings.Contains(err.Error(), "1048576") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
|
t.Fatalf("the inline declaration of more than 3 MiB: %v", err)
|
|
}
|
|
|
|
bucket, err := openBucket(ctx, conn, "novox")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
objects, err := byDigest.objects()
|
|
if err != nil || len(objects) != 3 {
|
|
t.Fatalf("objects: %d %v", len(objects), err)
|
|
}
|
|
if err := putObjects(ctx, bucket, objects); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := putObjects(ctx, bucket, objects); err != nil {
|
|
t.Fatalf("putting them again: %v", err)
|
|
}
|
|
refBody, err := byDigest.Body()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := link.Declare(ctx, bus, sigOf{}, "novox", refBody, 5*time.Second); err != nil {
|
|
t.Fatalf("the declaration by reference: %v", err)
|
|
}
|
|
|
|
// What the stream holds for the node is the reference form, and it names exactly the three objects.
|
|
latest, err := latestDeclarations(ctx, conn, []string{"novox", "ace"})
|
|
if err != nil || len(latest) != 1 {
|
|
t.Fatalf("the latest declarations: %v %v", len(latest), err)
|
|
}
|
|
named, err := namedByLatest("novox", latest["novox"])
|
|
if err != nil || len(named) != 3 {
|
|
t.Fatalf("what the latest declaration names: %v %v", named, err)
|
|
}
|
|
novoxBucket, err := broker.DeclarationFilesBucketOf("novox")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store, err := api.ObjectStore(ctx, novoxBucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var envelope struct {
|
|
Resources []sdkdecl.File `json:"resources"`
|
|
}
|
|
if err := json.Unmarshal(refBody, &envelope); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, f := range envelope.Resources {
|
|
if !f.IsReference() {
|
|
continue
|
|
}
|
|
name, _ := sdkdecl.ObjectName(f.Digest)
|
|
got, err := store.GetBytes(ctx, name)
|
|
if err != nil {
|
|
t.Fatalf("the object of %s: %v", f.ID, err)
|
|
}
|
|
if err := f.Verify(got); err != nil {
|
|
t.Errorf("the object does not verify: %v", err)
|
|
}
|
|
got[len(got)/2] ^= 1
|
|
if err := f.Verify(got); err == nil {
|
|
t.Errorf("a changed byte in the object of %s passed the verification", f.ID)
|
|
}
|
|
}
|
|
|
|
// The sweep against the real bucket: all named, nothing marked; a fourth object named by none is marked, kept a day less a
|
|
// minute, and removed at the day.
|
|
extra := bytes.Repeat([]byte("x"), 100)
|
|
extraName := strings.Repeat("e", 64)
|
|
if err := putObjects(ctx, bucket, map[string][]byte{extraName: extra}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b := busSweptBucket{store}
|
|
t0 := time.Now()
|
|
if r, err := sweepDeclarationFiles(ctx, b, named, t0); err != nil || r.Marked != 1 || r.Removed != 0 || r.Kept != 3 {
|
|
t.Fatalf("first sweep: %+v %v", r, err)
|
|
}
|
|
if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour-time.Minute)); err != nil || r.Removed != 0 {
|
|
t.Fatalf("a minute short of a day: %+v %v", r, err)
|
|
}
|
|
if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour+time.Second)); err != nil || r.Removed != 1 {
|
|
t.Fatalf("a day after: %+v %v", r, err)
|
|
}
|
|
if infos, err := store.List(ctx); err != nil || len(infos) != 3 {
|
|
t.Fatalf("the bucket holds %d objects (%v), want the 3 the declaration names", len(infos), err)
|
|
}
|
|
}
|
|
|
|
// 1696/AC8 on a real bucket: the sweep removes an object only if it still carries the mark it judged it by. An object marked
|
|
// unnamed, then put again by a send (which erases the mark), is kept when the sweep deletes it by the old mark.
|
|
func TestTheSweepsDeleteLeavesAnObjectASendPutAgainAfterItWasMarked(t *testing.T) { // 1696/AC8
|
|
s := testbus.Start(t)
|
|
conn, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
defer cancel()
|
|
bucket, err := openBucket(ctx, conn, "novox")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
name := strings.Repeat("a", 64)
|
|
content := map[string][]byte{name: bytes.Repeat([]byte("y"), 100)}
|
|
if err := putObjects(ctx, bucket, content); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b := busSweptBucket{bucket.store}
|
|
judged := time.Now().Add(-25 * time.Hour).Truncate(time.Second)
|
|
if err := b.MarkUnnamed(ctx, name, judged); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, marked, err := bucket.Has(ctx, name); err != nil || !marked {
|
|
t.Fatalf("the object was not marked: %v %v", marked, err)
|
|
}
|
|
// The sweep has read the mark; a send now puts the object again, which erases it.
|
|
if err := putObjects(ctx, bucket, content); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if have, marked, err := bucket.Has(ctx, name); err != nil || !have || marked {
|
|
t.Fatalf("a send did not put the marked object again: have %v marked %v %v", have, marked, err)
|
|
}
|
|
gone, err := b.Delete(ctx, name, judged)
|
|
if err != nil || gone {
|
|
t.Fatalf("the sweep deleted by an old mark: gone %v, %v", gone, err)
|
|
}
|
|
if have, _, err := bucket.Has(ctx, name); err != nil || !have {
|
|
t.Fatalf("the object a send relies on is gone: %v %v", have, err)
|
|
}
|
|
// And the mark still on it is honoured: marked again and deleted by that mark, it goes.
|
|
if err := b.MarkUnnamed(ctx, name, judged); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if gone, err := b.Delete(ctx, name, judged); err != nil || !gone {
|
|
t.Fatalf("an object still carrying the mark it was judged by is kept: gone %v, %v", gone, err)
|
|
}
|
|
}
|
|
|
|
// 1696/AC9 on a real bus: the sweep of a node's bucket takes the node's lock first, so it waits for a send that has put its objects
|
|
// and not yet published the declaration that names them.
|
|
func TestTheSweepOfANodesBucketHoldsTheNodesLock(t *testing.T) { // 1696/AC9
|
|
s := testbus.Start(t)
|
|
conn, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
defer cancel()
|
|
const node = "sweep-lock-node"
|
|
if _, err := openBucket(ctx, conn, node); err != nil { // the bucket exists, or the sweep would return before the lock
|
|
t.Fatal(err)
|
|
}
|
|
release := lockFilesOf(node) // a send is between its put and its publication
|
|
done := make(chan struct{})
|
|
go func() {
|
|
_, _ = sweepNode(ctx, conn, node)
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
t.Fatal("the sweep judged the node's bucket while a send held it")
|
|
case <-time.After(300 * time.Millisecond):
|
|
}
|
|
release()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(20 * time.Second):
|
|
t.Fatal("the sweep did not go on once the send let go")
|
|
}
|
|
}
|