diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index c9b95273..9b8e5c35 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -73,6 +73,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { return err } + if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil { + return err + } if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it", cred.EventConsumerName(), err) } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go index a2341544..6942aba5 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go @@ -69,3 +69,24 @@ func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline }() }) } + +// hearTakenIn starts the pipelines of each take-in.taken heard (§44, §54): one per entry, the message acknowledged once +// they are written, and heard again when a write failed, so the keys already written start nothing twice. +func hearTakenIn(hub *process.EventHub, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error { + return hub.Handle(pipeline.TakenInSubject, func(m jetstream.Msg) { + var ev catalog.TakeInTaken + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s is no take-in.taken: %v", m.Subject(), err) + _ = m.Ack() + return + } + n, err := starter.TakenIn(ev, v.View) + if err != nil { + logf("the take-in %s of %s: %d pipelines started, then %v; heard again", ev.Key, ev.Branch, n, err) + _ = m.Nak() + return + } + logf("the take-in %s of %s started %d pipelines", ev.Key, ev.Branch, n) + _ = m.Ack() + }) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index c507db63..e28f9c4a 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -134,3 +134,35 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou } return started, first } + +// TakenInSubject is where any module's take-in.taken is said, `mesh.mod..event.take-in.taken`: the runner names no +// module (§7), so the bus's grant from the manifest's `consumes` says which one it hears (§54). +const TakenInSubject = "mesh.mod.*.event.take-in.taken" + +// TakenIn starts, for a take-in.taken, one pipeline per entry of the source revision, keyed by its watched branch, that +// source revision and the module (design 47 G §44, §54, §476; ADR 0666 §13). A hand drop takes nothing in and starts +// nothing; a second start of an entry, a package-published having started it first, writes nothing (§54). view is the +// catalog half of the event's branch, read only for the revision its pipelines are made from. +func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half, bool)) (int, error) { + if ev.Dropped { + s.logf("the take-in %s was dropped by %s: it starts no pipeline", ev.Key, ev.By) + return 0, nil + } + if ev.Branch == "" || ev.Source <= 0 { + return 0, fmt.Errorf("a take-in.taken names its watched branch and source revision: key %q branch %q source %d", ev.Key, ev.Branch, ev.Source) + } + half, _ := view(ev.Branch) + started := 0 + var first error + for _, e := range ev.Entries { + ok, err := s.Start(Begin{Branch: ev.Branch, Source: ev.Source, Module: e.Module, Commit: e.Commit, Fp: e.Fingerprint, + Group: ev.Group, Repository: e.Repository, Revision: half.Revision}) + if err != nil && first == nil { + first = err + } + if ok { + started++ + } + } + return started, first +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go index 5b657f5b..32843100 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -85,3 +85,31 @@ func TestStarts(t *testing.T) { type failing struct{} func (failing) Create(string, []byte) error { return errors.New("bus away") } + +// A take-in.taken starts one pipeline per entry of its source revision (design 47 G §44, §54, §476); a hand drop starts +// none, and a publish that started an entry first leaves the take-in's start of it writing nothing. +func TestTakenInStarts(t *testing.T) { + store := created{} + var said []string + s := &Starter{Pipelines: store, Say: func(subject string, body any) error { said = append(said, subject); return nil }} + view := func(string) (Half, bool) { return Half{Revision: 12}, true } + ev := catalog.TakeInTaken{Kind: "taken", Branch: "main", Source: 7, Group: "g1", Entries: []catalog.TakeInEntry{ + {Module: "a", Commit: "c1", Fingerprint: "src2:aa", Repository: "o/a"}, {Module: "b", Commit: "c2", Fingerprint: "src2:bb"}}} + if _, err := s.Start(Begin{Branch: "main", Source: 7, Module: "b", Commit: "c2"}); err != nil { + t.Fatal(err) + } + n, err := s.TakenIn(ev, view) + if err != nil || n != 1 || len(said) != 2 { + t.Fatalf("a take-in started %d (%v), said %v: want a only, b being started already", n, err, said) + } + var p seat.Pipeline + if err := json.Unmarshal(store["main.7.a"], &p); err != nil || p.Commit != "c1" || p.Group != "g1" || p.Revision != 12 || p.Fingerprint != "src2:aa" { + t.Errorf("the pipeline main.7.a = %+v (%v)", p, err) + } + if n, err := s.TakenIn(catalog.TakeInTaken{Kind: "taken", Dropped: true, By: "op", Branch: "main", Source: 8, Entries: ev.Entries}, view); n != 0 || err != nil { + t.Errorf("a hand drop started %d (%v)", n, err) + } + if _, err := s.TakenIn(catalog.TakeInTaken{Kind: "taken", Branch: "main"}, view); err == nil { + t.Error("a take-in naming no source revision is refused") + } +}