diff --git a/cmd/mesh-controller/declaration_capability_integration_test.go b/cmd/mesh-controller/declaration_capability_integration_test.go new file mode 100644 index 00000000..f408079c --- /dev/null +++ b/cmd/mesh-controller/declaration_capability_integration_test.go @@ -0,0 +1,110 @@ +//go:build integration + +package main + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/overlay" + + "git.novox.be/novox/mesh-sdk/go/compose" +) + +// 1696/AC2, ADR 0486 ยง5: a node is sent a file by digest only when its profile reports the capability. A node whose profile lacks +// it, and a node the mesh has no profile of, are sent every content inline; a node that reports it is sent the reference. The test +// goes through composeQuestion and composeSend, the two places readsFileReferences is read (compose_sdk.go). +func TestAFileTravelsByDigestOnlyToANodeThatReportsTheCapability(t *testing.T) { + open := aMesh(t) // anchor and laptop report no file-references + ctx := t.Context() + big := strings.Repeat("0123456789abcdef", 8<<10) // 128 KiB + register(t, open, catalogue.Manifest{Module: "bigfile", Version: "1", + Resources: []map[string]any{{"id": "blob", "type": "file", "path": "/etc/bigfile/blob", "mode": "0644", "content": big}}}) + + // reader reports the capability; bare has never reported a profile at all. + for i, n := range []struct { + name string + profile map[string]any + }{ + {"reader", map[string]any{"capabilities": []any{ + map[string]any{"name": "container-runtime", "present": true}, + map[string]any{"name": "systemd", "present": true}, + map[string]any{"name": sdkdecl.FileReferencesCapability, "present": true}}}}, + {"bare", nil}, + } { + record, err := open.inventory.AddNode(ctx, n.name) + if err != nil { + t.Fatal(err) + } + if err := open.inventory.SetPlace(ctx, n.name, n.name+".example:51820", "here", false, fmt.Sprintf("10.77.0.%d", i+10)); err != nil { + t.Fatal(err) + } + if n.profile != nil { + if err := open.inventory.RecordProfile(ctx, record.ID, n.profile); err != nil { + t.Fatal(err) + } + } + if err := open.inventory.RecordSealingKey(ctx, record.ID, aPublicKey(t)); err != nil { + t.Fatal(err) + } + if err := open.inventory.RecordOverlayKey(ctx, record.ID, aPublicKey(t)); err != nil { + t.Fatal(err) + } + if _, err := open.inventory.Assign(ctx, n.name, overlay.Name); err != nil { + t.Fatal(err) + } + } + for _, node := range []string{"laptop", "reader", "bare"} { + if _, err := assign(ctx, open, node, "bigfile"); err != nil { + t.Fatalf("assign %s: %v", node, err) + } + } + + byDigest := map[string]bool{"laptop": false, "bare": false, "reader": true} + for node, want := range byDigest { + asked, _, err := composeQuestion(ctx, open, node, compose.Reading, false) + if err != nil { + t.Fatalf("%s: %v", node, err) + } + sent, err := composeSend(ctx, open, node) + if err != nil { + t.Fatalf("%s: %v", node, err) + } + for what, s := range map[string]sendable{"question": asked, "send": sent} { + if s.fileReferences != want { + t.Fatalf("%s, %s: fileReferences %v, want %v", node, what, s.fileReferences, want) + } + body, err := s.Body() + if err != nil { + t.Fatal(err) + } + var got struct { + Resources []map[string]any `json:"resources"` + } + if err := json.Unmarshal(body, &got); err != nil { + t.Fatal(err) + } + var found bool + for _, r := range got.Resources { + if r["path"] != "/etc/bigfile/blob" { + continue + } + found = true + inline := r["content"] == big + if want && (inline || r["digest"] == nil || r["size"] == nil) { + t.Fatalf("%s, %s: a node that reads references is sent the content inline", node, what) + } + if !want && (!inline || r["digest"] != nil) { + t.Fatalf("%s, %s: a node that does not report the capability is sent a reference", node, what) + } + } + if !found { + t.Fatalf("%s, %s: the file is not in the declaration", node, what) + } + } + } +} diff --git a/cmd/mesh-controller/declaration_files_integration_test.go b/cmd/mesh-controller/declaration_files_integration_test.go index b5b45baf..b554e115 100644 --- a/cmd/mesh-controller/declaration_files_integration_test.go +++ b/cmd/mesh-controller/declaration_files_integration_test.go @@ -159,3 +159,88 @@ func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) 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") + } +} diff --git a/cmd/mesh-controller/declaration_files_test.go b/cmd/mesh-controller/declaration_files_test.go index 728ad11e..c428d418 100644 --- a/cmd/mesh-controller/declaration_files_test.go +++ b/cmd/mesh-controller/declaration_files_test.go @@ -467,3 +467,68 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi t.Fatal("the count did not start again after a put that worked") } } + +// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held. +type aBusThatSays struct { + link.Bus + published chan string + heldAtPublish func() bool +} + +func (b *aBusThatSays) PublishDeclaration(_ context.Context, node string, _ []byte) error { + if b.heldAtPublish != nil && !b.heldAtPublish() { + return errors.New("published while the node's bucket was not held") + } + b.published <- node + return errors.New("stop here: nothing is recorded") +} + +type aSignerOfZeros struct{} + +func (aSignerOfZeros) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil } + +// whileHeld says whether the node's lock is held by someone right now, without keeping it. +func whileHeld(node string) bool { + got := make(chan struct{}) + go func() { + defer lockFilesOf(node)() + close(got) + }() + select { + case <-got: + return false + case <-time.After(100 * time.Millisecond): + // held: the goroutine stays waiting and takes and releases the lock once it is let go. + return true + } +} + +// 1696/AC9: a send holds the node's lock from the put to the publication, so the sweep never removes an object between them. The +// test goes through overTheBus.declare: while another holds the lock the declaration is not published, and at the publication the +// lock is held. +func TestASendHoldsTheNodesLockFromThePutToThePublication(t *testing.T) { // 1696/AC9 + const node = "lock-send-node" + bus := &aBusThatSays{published: make(chan string, 1)} + bus.heldAtPublish = func() bool { return whileHeld(node) } + send := overTheBus{server: link.SendingOn(bus), signer: aSignerOfZeros{}} + release := lockFilesOf(node) // a sweep is judging this node's bucket + done := make(chan error, 1) + go func() { + _, err := send.declare(t.Context(), readyNode{node: node}, []byte(`{}`)) + done <- err + }() + select { + case n := <-bus.published: + t.Fatalf("%s was sent its declaration while the sweep held the node's bucket", n) + case <-time.After(150 * time.Millisecond): + } + release() + select { + case <-bus.published: + case <-time.After(5 * time.Second): + t.Fatal("the send did not go on once the sweep let go") + } + if err := <-done; err == nil || !strings.Contains(err.Error(), "nothing is recorded") { + t.Fatalf("the send did not reach the publication (a lock held at it): %v", err) + } +} diff --git a/internal/link/bus_nats.go b/internal/link/bus_nats.go index bfaebdb4..2aebcb1d 100644 --- a/internal/link/bus_nats.go +++ b/internal/link/bus_nats.go @@ -24,3 +24,6 @@ func ConnectNats(js *broker.JetStream, enroller Enroller, listener Listener) *Se // Bus is the controller's outbound, whichever transport it connected over. Callers that send a // declaration or ask a tool use this rather than the channel, which one transport does not have. func (s *Server) Bus() Bus { return s.bus } + +// SendingOn is a server that only sends, over the bus given: for a test of a sender with a bus it holds in its hand. +func SendingOn(bus Bus) *Server { return &Server{bus: bus} }