Artifact [02f0b45083]

Artifact 02f0b450838bcbef08f250dbbc6270cf7b136289ca9e5c808fc9c6729775462a:

• File internal/ci/trigger.go — part of check-in [0502a93834] at 2026-08-17 13:02:21 on branch trunk — ci: the trigger half of v3 — dagnats is the engine

  0001 §6 splits this narrowly: dagnats runs pipelines, quarry notices a check-in and shows the result. quarry defines nothing, schedules nothing, retries nothing.

  A pipeline is TRACKED CONTENT — ci/*.json in the repository, dagnats' native format — read at the check-in under test, so the pipeline for a historical check-in is the pipeline that ran and there is no forge-side store to drift. Only files directly under ci/ count; a nested path is somebody's fixture.

  Two behaviours worth naming. First sight of a repository takes the tip as a watermark and launches NOTHING: running a pipeline per historical check-in would be a denial of service against the engine and would say nothing about code nobody just changed — on an imported corpus that is tens of thousands of runs. And an unrecognised run status counts as still running, because claiming a run finished when it has not is how a verdict gets written against a build still going.

  Verified against the live dagnats for the wire shapes; the trigger itself is covered by tests. (user: dmestas, size: 3491) [annotate] [blame] [check-ins using]

package ci

import (
	"fmt"
	"log"
)

// Store is the record-keeping the trigger needs; the OLTP database
// satisfies it.
type Store interface {
	CIWatermark(repo string) (string, bool)
	SetCIWatermark(repo, checkin string) error
	AddRun(runID, repo, checkin, workflow, pipeline, status string) error
}

// Repo is the read side the trigger needs: which check-ins are new, and
// what the repository's pipelines look like AT that check-in.
type Repo interface {
	// NewCheckins returns check-ins after the watermark, oldest first.
	// An empty watermark means "the tip only" — a first run must not
	// launch a pipeline for every check-in in a repository's history,
	// which on an imported corpus would be tens of thousands.
	NewCheckins(after string, limit int) ([]Checkin, error)
	// Files lists a check-in's paths.
	Files(version string) ([]string, error)
	// Read returns one file's bytes at a check-in.
	Read(path, version string) ([]byte, error)
}

// Checkin is the context a pipeline receives.
type Checkin struct {
	Hash    string
	Branch  string
	Date    string
	User    string
	Comment string
}

// Trigger launches pipelines for check-ins a repository has not seen.
// It is deliberately quiet about the ordinary case — most repositories
// have no ci/ directory, and that is not a condition worth logging on
// every sync round.
func Trigger(e *Engine, st Store, repo string, r Repo, cloneURL string) error {
	if !e.Enabled() {
		return nil
	}
	mark, ok := st.CIWatermark(repo)
	if !ok {
		// First sight of this repository: take the tip as the watermark
		// without running anything. Launching the whole history would
		// be a denial of service against the engine and would say
		// nothing useful about code nobody just changed.
		tip, err := r.NewCheckins("", 1)
		if err != nil || len(tip) == 0 {
			return err
		}
		return st.SetCIWatermark(repo, tip[len(tip)-1].Hash)
	}
	fresh, err := r.NewCheckins(mark, 20)
	if err != nil {
		return err
	}
	for _, c := range fresh {
		if err := launch(e, st, repo, r, c, cloneURL); err != nil {
			// One bad pipeline must not wedge the watermark: the next
			// check-in still gets its chance, and the failure is named.
			log.Printf("ci: %s %.12s: %v", repo, c.Hash, err)
		}
		if err := st.SetCIWatermark(repo, c.Hash); err != nil {
			return err
		}
	}
	return nil
}

func launch(e *Engine, st Store, repo string, r Repo, c Checkin, cloneURL string) error {
	files, err := r.Files(c.Hash)
	if err != nil {
		return err
	}
	var pipelines []string
	for _, f := range files {
		if IsPipeline(f) {
			pipelines = append(pipelines, f)
		}
	}
	if len(pipelines) == 0 {
		return nil // no ci/ directory: the common case, silently fine
	}
	for _, p := range pipelines {
		def, err := r.Read(p, c.Hash)
		if err != nil {
			return fmt.Errorf("read %s: %w", p, err)
		}
		// Registered from the definition AT THIS CHECK-IN, so a pipeline
		// edit takes effect for the check-in that made it and not before.
		wf, err := e.Register(def)
		if err != nil {
			return err
		}
		runID, err := e.Start(wf.Name, map[string]any{
			"repo":     repo,
			"checkin":  c.Hash,
			"branch":   c.Branch,
			"date":     c.Date,
			"user":     c.User,
			"comment":  c.Comment,
			"cloneUrl": cloneURL,
			"pipeline": p,
		})
		if err != nil {
			return err
		}
		if err := st.AddRun(runID, repo, c.Hash, wf.Name, p, "running"); err != nil {
			return err
		}
		log.Printf("ci: %s %.12s -> %s run %s", repo, c.Hash, wf.Name, runID)
	}
	return nil
}