RN3 (take-in half): the runner starts one pipeline per entry of a take-in.taken, a second start writing nothing

Builds: design 47 G §44, §54, §103, §476 (recovers the start of d8dfc557 in the form the design has today)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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)
 	}

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()
+	})
+}

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.<module>.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
+}

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")
+	}
+}