diff --git a/cmd/mesh-host/keptartifact.go b/cmd/mesh-host/keptartifact.go new file mode 100644 index 00000000..0e7dcdb0 --- /dev/null +++ b/cmd/mesh-host/keptartifact.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode" + "github.com/novox/mesh-host/internal/apply" +) + +// The artifact store's holder takes its own new build from the build node that kept it (design 47): its done build names the +// node and the digest, the node's build seat answers where the bytes are as an object on the bus, and the node-engine +// reads them and checks them against the digest. Never another module's files: the seat's verb is the way. + +// keptAskWithin bounds one ask of a build seat; keptObjectWithin one read of the object. +var ( + keptAskWithin = 30 * time.Second + keptObjectWithin = 10 * time.Minute +) + +// keptBuildSeats are the three build seats, each of which serves the verb kept-artifact: the node that kept the build +// holds the seat of the language it was built in, and the others do not answer. +var keptBuildSeats = []string{nodebuildgo.Seat, nodebuildnode.Seat, nodebuildimage.Seat} + +// keptArtifactFetch is the apply's fetch of what a module's done build kept on a build node. +func keptArtifactFetch(buildOf func(context.Context, string, string) (string, string, bool, error), + ask func(ctx context.Context, node, seat, verb string, args any, within time.Duration) (json.RawMessage, error), + object func(ctx context.Context, store, name string, limit int64) ([]byte, error)) apply.KeptFetch { + return func(ctx context.Context, module, digest string) ([]byte, bool, error) { + node, fp, kept, err := buildOf(ctx, module, digest) + if err != nil || !kept { + return nil, false, err + } + hex := strings.TrimPrefix(digest, "sha256:") + var notHeld []string + for _, seat := range keptBuildSeats { + raw, err := ask(ctx, node, seat, nodebuildgo.VerbKeptArtifact, + nodebuildgo.KeptArtifactInput{Module: module, Fp: fp, Digest: hex}, keptAskWithin) + if err != nil { + notHeld = append(notHeld, seat+": "+err.Error()) + continue + } + var got nodebuildgo.KeptArtifactAnswer + if json.Unmarshal(raw, &got) != nil { + return nil, true, fmt.Errorf("%s of %s answered with something that is not an answer of kept-artifact", seat, node) + } + switch { + case got.Error != "": + return nil, true, fmt.Errorf("%s of %s could not give the kept build of %s: %s", seat, node, module, got.Error) + case got.Refused != "": + return nil, true, fmt.Errorf("%s of %s refuses the kept build of %s: %s %s", seat, node, module, got.Refused, got.Words) + case got.Store == "" || got.Object == "" || got.Digest != hex: + return nil, true, fmt.Errorf("%s of %s answered for %s without the store and the object of the digest %s", seat, node, module, hex) + } + octx, cancel := context.WithTimeout(ctx, keptObjectWithin) + body, err := object(octx, got.Store, got.Object, got.Size+1) + cancel() + if err != nil { + return nil, true, fmt.Errorf("the build of %s kept on %s could not be read from %s: %w", module, node, got.Store, err) + } + return body, true, nil + } + return nil, true, fmt.Errorf("%s keeps the build of %s for the artifact store and none of its build seats answers (%s)", + node, module, strings.Join(notHeld, "; ")) + } +} diff --git a/cmd/mesh-host/keptartifact_test.go b/cmd/mesh-host/keptartifact_test.go new file mode 100644 index 00000000..b662f01b --- /dev/null +++ b/cmd/mesh-host/keptartifact_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" +) + +// A module's done build that kept its artifact on a build node is fetched through that node's build seat (design 47). +func TestAKeptArtifactIsFetchedThroughTheBuildSeatOfItsNode(t *testing.T) { + hex := strings.Repeat("a", 64) + stored := func(string) (json.RawMessage, error) { + return json.RawMessage(`{"module":"distribution","fp":"f1","digest":"` + hex + `","size":5,"store":"kept-builds","object":"` + hex + `"}`), nil + } + for _, row := range []struct { + name string + kept bool + answer func(seat string) (json.RawMessage, error) + object error + wantKept bool + wantErr string + wantSeat string + }{ + {"a build kept nothing is not asked", false, nil, nil, false, "", ""}, + {"the seat of the language answers", true, func(seat string) (json.RawMessage, error) { + if seat != "node-build-node" { + return nil, errors.New("no responder") + } + return stored(seat) + }, nil, true, "", "node-build-node"}, + {"no build seat of the node answers", true, func(string) (json.RawMessage, error) { return nil, errors.New("no responder") }, + nil, true, "novox keeps the build of distribution", ""}, + {"the seat refuses it", true, func(string) (json.RawMessage, error) { + return json.RawMessage(`{"module":"distribution","fp":"f1","refused":"not kept","words":"swept"}`), nil + }, nil, true, "refuses the kept build of distribution: not kept", ""}, + {"the object cannot be read", true, stored, errors.New("no object"), true, "could not be read from kept-builds", "node-build-go"}, + } { + t.Run(row.name, func(t *testing.T) { + var seats []string + fetch := keptArtifactFetch( + func(context.Context, string, string) (string, string, bool, error) { return "novox", "f1", row.kept, nil }, + func(_ context.Context, node, seat, verb string, _ any, _ time.Duration) (json.RawMessage, error) { + if node != "novox" || verb != "kept-artifact" { + t.Errorf("asked %s of %s on %s", verb, seat, node) + } + seats = append(seats, seat) + return row.answer(seat) + }, + func(_ context.Context, store, name string, limit int64) ([]byte, error) { + if store != "kept-builds" || name != hex || limit < 5 { + t.Errorf("read %s/%s up to %d", store, name, limit) + } + return []byte("bytes"), row.object + }) + body, kept, err := fetch(context.Background(), "distribution", "sha256:"+hex) + if kept != row.wantKept { + t.Fatalf("kept = %v", kept) + } + if row.wantErr == "" && (err != nil || (kept && string(body) != "bytes")) { + t.Fatalf("got %q, %v", body, err) + } + if row.wantErr != "" && (err == nil || !strings.Contains(err.Error(), row.wantErr)) { + t.Fatalf("the error is %v, want %q", err, row.wantErr) + } + if !row.kept && len(seats) > 0 { + t.Errorf("a build that kept nothing asked %v", seats) + } + }) + } +} diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 2b5b8b9b..45d0ba7d 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -448,6 +448,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus } began := time.Now() node := session.Membership.Node + var keptFetch atomic.Pointer[apply.KeptFetch] // the composer's, set once it exists composedBucket, err := compositions.BucketFor(node) if err != nil { cancel() @@ -550,6 +551,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus Backoff: link.Backoff, Apply: func(ctx context.Context, c compositions.Composed, others []compositions.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) + if kf := keptFetch.Load(); kf != nil { + actx = apply.WithKept(actx, *kf) + } // While the grants part's list is followed, the controller's list in the bus module's composition is not // written: the apply keeps the list on the machine (ADR 0547 clause 8, never both). held, herr := users.Hold(c) @@ -681,6 +685,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus apply.UseEngineModule(composer.EngineModule) // the guard of the engine's own trees fixes no name (§527) env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting + keptFetchFn := keptArtifactFetch(composer.KeptBuildOf, session.AskSeatOn, session.KeptObject) + keptFetch.Store(&keptFetchFn) env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does session.NoteComposes(linked, nodeengineclient.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere") ag = agent.New(env) diff --git a/go.mod b/go.mod index 04fc37be..a5028435 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,9 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate v0.0.0-00010101000000-000000000000 diff --git a/internal/apply/fetch_first.go b/internal/apply/fetch_first.go index ad91ec02..e4645f80 100644 --- a/internal/apply/fetch_first.go +++ b/internal/apply/fetch_first.go @@ -60,10 +60,21 @@ func fetchFirst(ctx context.Context, ordered []declaration.Resource, d *declarat continue } } - got, seen := bySource[source] - if !seen { + // What the module's done build kept on a build node is fetched from there, not from the store (design 47). + cacheKey, kept := source, false + if kf := keptFetchFrom(ctx); kf != nil && ours { + cacheKey = "kept " + module + " " + digest + if _, seen := bySource[cacheKey]; seen { + kept = true + } else if body, isKept, err := kf(ctx, module, digest); isKept { + kept = true + bySource[cacheKey] = fetchedArtifact{body: body, err: err} + } + } + got, seen := bySource[cacheKey] + if !seen && !kept { got.body, got.err = fetchMinding(ctx, source, in.away) - bySource[source] = got + bySource[cacheKey] = got } if got.err == nil { sum := sha256.Sum256(got.body) diff --git a/internal/apply/kept_artifact.go b/internal/apply/kept_artifact.go new file mode 100644 index 00000000..3585470a --- /dev/null +++ b/internal/apply/kept_artifact.go @@ -0,0 +1,29 @@ +package apply + +import "context" + +// The build the artifact store's holder keeps on a build node (novox/hq design 47, *The artifact store's holder takes its +// own new build even when its seat is the one broken*): the holder's program cannot be fetched from the store it serves, so +// its done build names the node that keeps it and the node-engine fetches it from there, through the build seat's verb. +// How is the caller's (a hook on the apply's context, as the files' link is); what apply does is use the bytes in place of +// the source's fetch, check them against the declared digest as every fetch is, and gate the module when they cannot be had. + +// KeptFetch gives the bytes of the artifact a module's done build kept on a build node, by the artifact's digest +// ("sha256:" and 64 hex digits, as a declaration names it). kept is false when the module's build kept no such artifact: +// the artifact is then fetched from its source, as before. An error says the artifact is kept and could not be had. +type KeptFetch func(ctx context.Context, module, digest string) (body []byte, kept bool, err error) + +type keptFetchKey struct{} + +// WithKept is ctx carrying how a kept artifact is fetched. +func WithKept(ctx context.Context, f KeptFetch) context.Context { + if f == nil { + return ctx + } + return context.WithValue(ctx, keptFetchKey{}, f) +} + +func keptFetchFrom(ctx context.Context) KeptFetch { + f, _ := ctx.Value(keptFetchKey{}).(KeptFetch) + return f +} diff --git a/internal/apply/replace_image_test.go b/internal/apply/replace_image_test.go index fa48c838..a7fd05d5 100644 --- a/internal/apply/replace_image_test.go +++ b/internal/apply/replace_image_test.go @@ -164,6 +164,47 @@ func TestARunningContainerIsKeptWhenItsReplacementImageCannotBeObtained(t *testi t.Errorf("the container's outcome is %+v, want skipped", o) } }) + + // Design 47 (W13.8): the holder's own program is fetched from the build node that kept it, not from the store; when it + // cannot be had, or is not the declared digest, nothing of the module is replaced and the failure says so. + for _, row := range []struct { + name string + fetch KeptFetch + wantErr string + }{ + {"the node that keeps it is away", func(context.Context, string, string) ([]byte, bool, error) { + return nil, true, errors.New("novox keeps it and does not answer") + }, "novox keeps it and does not answer"}, + {"what the node kept is not the declared digest", func(context.Context, string, string) ([]byte, bool, error) { + return []byte("other"), true, nil + }, "what arrived is"}, + } { + t.Run("a kept artifact: "+row.name, func(t *testing.T) { + gone := httptest.NewServer(http.NotFoundHandler()) + source := gone.URL + "/program.tar.gz" + gone.Close() + was := storeAwayAtMost + storeAwayAtMost = 0 + t.Cleanup(func() { storeAwayAtMost = was }) + d := parseTrusted(t, `{"declaration":1,"resources":[ + {"id":"distribution.program","type":"process","name":"program","source":"`+source+`", + "digest":"sha256:`+strings.Repeat("0", 64)+`","run":["program"]} + ]}`) + var asked []string + fetch := func(ctx context.Context, module, digest string) ([]byte, bool, error) { + asked = append(asked, module+" "+digest) + return row.fetch(ctx, module, digest) + } + run := func(ctx context.Context, name string, args ...string) (string, error) { return "", nil } + _, _, err := Apply(WithKept(context.Background(), fetch), archHost(t), d, store.State{}, store.OriginCarried, run, nil, nil) + if err == nil || !strings.Contains(err.Error(), row.wantErr) { + t.Errorf("the failure is %v, want it to say %q", err, row.wantErr) + } + if len(asked) != 1 || asked[0] != "distribution sha256:"+strings.Repeat("0", 64) { + t.Errorf("the kept fetch was asked %q", asked) + } + }) + } } func TestAReplacementImageIsPresentBeforeTheRunningContainerIsRemoved(t *testing.T) { diff --git a/internal/link/files.go b/internal/link/files.go index 44d02ddf..23f24eaa 100644 --- a/internal/link/files.go +++ b/internal/link/files.go @@ -172,6 +172,49 @@ func (l *natsLink) GrantsObject(ctx context.Context, name string, limit int64) ( return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.GrantsObject(ctx, name, limit) } +// KeptObject reads one object of a build module's store of kept builds by its name (design 47: the build the artifact store's +// holder keeps on a build node; the build seat's verb kept-artifact names the store and the object), with the same +// subjects as FileObject, verified against its digest. +func (b OverNATS) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + js := b.JS + if b.Conn != nil { + wait := 10 * time.Second + if at, ok := ctx.Deadline(); ok { + if wait = time.Until(at); wait <= 0 { + return nil, context.DeadlineExceeded + } + } + bounded, err := b.Conn.JetStream(nats.MaxWait(wait)) + if err != nil { + return nil, err + } + js = bounded + } + return objectByName(ctx, js, store, declaration.FileStreamPrefix+store, name, limit) +} + +func (l *natsLink) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.KeptObject(ctx, store, name, limit) +} + +// KeptObjects is a link that can read an object of a build module's store of kept builds. +type KeptObjects interface { + KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) +} + +// KeptObject reads an object of a build module's store of kept builds on the link open now. +func (q *Session) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + k, ok := bus.(KeptObjects) + if bus == nil || !ok { + return nil, ErrNoLink + } + return k.KeptObject(ctx, store, name, limit) +} + // GrantsObjects is a link that can read an object of the grants part's bucket of sealed user lists. type GrantsObjects interface { GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) diff --git a/internal/link/session.go b/internal/link/session.go index 2d74daf2..59bfb062 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -348,6 +348,12 @@ func SeatToolSubject(seat, verb, node string) string { // AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal // as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14). func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { + return q.AskSeatOn(ctx, q.Membership.Node, seat, verb, args, within) +} + +// AskSeatOn asks the holder of a node seat on another node a verb: the build seat of the node a want's `kept-on` names +// gives the kept build (design 47 G section 526). +func (q *Session) AskSeatOn(ctx context.Context, node, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { q.init() q.mu.Lock() bus := q.bus @@ -365,7 +371,7 @@ func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, with } asking, cancel := context.WithTimeout(ctx, within) defer cancel() - reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body) + reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, node), body) if err != nil { return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err) } diff --git a/internal/selfcompose/kept.go b/internal/selfcompose/kept.go new file mode 100644 index 00000000..52abee8c --- /dev/null +++ b/internal/selfcompose/kept.go @@ -0,0 +1,31 @@ +package selfcompose + +import ( + "context" + "strings" +) + +// KeptBuildOf is the node that keeps the build a module is wanted at, and its fingerprint, for the artifact store's +// holder (design 47 G section 271, 526, A-50): the want's `kept-on` and nothing else names it, the builder's builds being +// no read of the node-engine's (section 421). kept is false when the module has no want, its want names no keeping node, +// or its `parts` do not name the digest ("sha256:" and 64 hex digits, or bare): the artifact is then fetched from the store. +func (c *Composer) KeptBuildOf(_ context.Context, module, digest string) (node, fp string, kept bool, err error) { + c.mu.Lock() + w, ok := c.wants[module] + c.mu.Unlock() + if !ok || w.Fp == "" || w.KeptOn == "" { + return "", "", false, nil + } + if len(w.Parts) > 0 { + named := false + for _, d := range w.Parts { + if strings.TrimPrefix(d, "sha256:") == strings.TrimPrefix(digest, "sha256:") { + named = true + } + } + if !named { + return "", "", false, nil + } + } + return w.KeptOn, w.Fp, true, nil +} diff --git a/internal/selfcompose/kept_test.go b/internal/selfcompose/kept_test.go new file mode 100644 index 00000000..c2b7f305 --- /dev/null +++ b/internal/selfcompose/kept_test.go @@ -0,0 +1,35 @@ +package selfcompose + +import ( + "context" + "strings" + "testing" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// The node that keeps a kept build is the want's kept-on alone, for an artifact its parts name (design 47 G 526, A-50). +func TestTheKeepingNodeIsTheWantsKeptOn(t *testing.T) { + kd := strings.Repeat("6", 64) + for _, row := range []struct { + name, digest string + want *meshwants.WantsByWantNodeModuleValue + node string + kept bool + }{ + {"kept, by its bare digest", kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "n3", true}, + {"kept, by the declared digest", "sha256:" + kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "n3", true}, + {"a digest the parts do not name", "sha256:" + strings.Repeat("7", 64), &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "", false}, + {"a want naming no keeping node", kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", Parts: map[string]string{"db:p": "sha256:" + kd}}, "", false}, + {"no want of the module", kd, nil, "", false}, + } { + c := &Composer{wants: map[string]meshwants.WantsByWantNodeModuleValue{}} + if row.want != nil { + c.wants["db"] = *row.want + } + node, fp, kept, err := c.KeptBuildOf(context.Background(), "db", row.digest) + if err != nil || node != row.node || kept != row.kept || (kept && fp != "f1") { + t.Errorf("%s: node %q fp %q kept %v err %v", row.name, node, fp, kept, err) + } + } +}