vole

Clone

Read only, no credentials:

https://rodent.software/git/vole.git

Write over SSH with a registered key:

git clone ssh://git@0.0.0.0:2222/vole.git

A free and open-source forge that lives in your Git repo.

Blame · server/ssh/runner.go

Line attribution on main · 639 line(s)

View file History

c5902451 Keenan Nemetz 16d ago	1	package ssh
b44cbfaf Keenan Nemetz 17d ago	2
	3	import (
	4	"encoding/json"
78cd3437 Keenan Nemetz 12d ago	5	"errors"
213f4d6d Keenan Nemetz 17d ago	6	"fmt"
78cd3437 Keenan Nemetz 12d ago	7	"io"
b44cbfaf Keenan Nemetz 17d ago	8	"time"
b44cbfaf Keenan Nemetz 17d ago	9
c5902451 Keenan Nemetz 16d ago	10	xssh "golang.org/x/crypto/ssh"
d37d301f Keenan Nemetz 15h ago	11	"rodent.software/vole/ci/runs"
	12	"rodent.software/vole/core/id"
	13	"rodent.software/vole/core/lfs"
b44cbfaf Keenan Nemetz 17d ago	14	)
	15
	16	// The runner protocol.
	17	//
	18	// A runner connects over the same SSH transport as everything else, so it
	19	// authenticates with a key and needs no token: the account it connects as decides
	20	// which repositories it may run jobs for. One JSON object per line, one reply per
	21	// request, on a single channel.
	22	//
90fdaf39 Keenan Nemetz 16d ago	23	// {"op":"lease"} -> {"run":{...}} or {}
90fdaf39 Keenan Nemetz 16d ago	24	// {"op":"heartbeat","run":"<id>"} -> {"ok":true\|false}
b312e166 Keenan Nemetz 10d ago	25	// {"op":"finish","run":"<id>","record":{...}} -> {"ok":true\|false}
90fdaf39 Keenan Nemetz 16d ago	26	//
b312e166 Keenan Nemetz 10d ago	27	// A finish carries the attempt's record: the runner uploads its log and artifacts,
	28	// then sends the record on this channel. The server checks that it names the
	29	// account that holds the lease, that it describes the job the queue handed out, and
	30	// that every log and artifact pointer exists in the object store. It then stores the
	31	// attempt and the run's result in one transaction. The record is not a git commit;
	32	// CI history is instance-local.
	33	//
	34	// A tagged build that publishes release artifacts still pushes the release pointers
	35	// to the metadata branch after the record is stored. Releases are product metadata,
	36	// not CI history.
b44cbfaf Keenan Nemetz 17d ago	37	//
b44cbfaf Keenan Nemetz 17d ago	38	// "ok":false means the lease is gone — another runner has the job — and the runner
b312e166 Keenan Nemetz 10d ago	39	// must stop rather than report a result nobody will record. A retry of the same
	40	// record after a lost reply is answered "ok":true, because the bytes are already
	41	// stored and the attempt is append-only.
b44cbfaf Keenan Nemetz 17d ago	42
34afc569 Keenan Nemetz 16d ago	43	const (
	44	// eligibleFor is how long a session reuses its repository list. Access rules live
	45	// in the instance database and only change when an admin edits them, so a runner
	46	// re-derives its list on the roster's five-second beat instead of on every poll.
	47	eligibleFor = 5 * time.Second
	48	// maxLeaseWait caps what a runner may ask for. The wait is a hint — the lease
	49	// remains the database's decision — so all a long wait holds is one SSH channel.
	50	maxLeaseWait = 60 * time.Second
b312e166 Keenan Nemetz 10d ago	51	// maxRunnerRequestBytes bounds one request line. A lease is small, but a finish
	52	// carries the attempt's steps and artifact names, so the bound has room for a
	53	// run at the artifact cap. It is per line, so a long-lived runner is not limited
	54	// in total.
	55	maxRunnerRequestBytes = 4 << 20
34afc569 Keenan Nemetz 16d ago	56	)
34afc569 Keenan Nemetz 16d ago	57
78cd3437 Keenan Nemetz 12d ago	58	// errRequestTooLarge ends a runner session whose request line outgrew the bound.
	59	var errRequestTooLarge = errors.New("runner request line is too large")
	60
	61	// lineLimitReader fails once a single newline-terminated line passes limit bytes.
	62	// It counts bytes as they cross the reader, so json.Decoder's internal buffering
	63	// cannot hide a large request, and it resets at every newline so one oversized
	64	// line is the only thing it refuses.
	65	type lineLimitReader struct {
	66	r io.Reader
	67	limit int64
	68	line int64
	69	}
	70
	71	func (l *lineLimitReader) Read(p []byte) (int, error) {
	72	n, err := l.r.Read(p)
	73	for _, b := range p[:n] {
	74	if b == '\n' {
	75	l.line = 0
	76	continue
	77	}
	78	l.line++
	79	if l.line > l.limit {
	80	return 0, errRequestTooLarge
	81	}
	82	}
	83	return n, err
	84	}
	85
b44cbfaf Keenan Nemetz 17d ago	86	type runnerRequest struct {
90fdaf39 Keenan Nemetz 16d ago	87	Op string `json:"op"`
90fdaf39 Keenan Nemetz 16d ago	88	Run string `json:"run"`
b238ac24 Keenan Nemetz 16d ago	89	// Repos narrows this session's leases to the repositories the runner was started
	90	// for (--repos). It is applied as an intersection with what the account's runner
	91	// allowlists grant, so it can only subtract: a runner cannot ask its way into a
	92	// repository no allowlist names. Empty means "no narrowing", which keeps the list
	93	// following allowlist edits instead of freezing it at connect time.
	94	Repos []string `json:"repos,omitempty"`
6c03e461 Keenan Nemetz 16d ago	95	// Labels are what this machine is (--label), and a job is only leased to a runner
	96	// carrying every label the job asked for with runs-on. They describe a machine, not
	97	// a permission: a runner that claims a label it does not have only hurts itself,
	98	// which is why this is a client claim where --repos is server-enforced.
	99	Labels []string `json:"labels,omitempty"`
d7958024 Keenan Nemetz 6d ago	100	// Network is the network this machine can give, derived from --container-network.
	101	// A job that declares `network: bridge` is only leased to a runner that says it can
	102	// provide one, so the job waits for a capable machine instead of running without
	103	// the network it asked for. A runner that claims more than it can give only fails
	104	// its own jobs, which is the same trade as a label.
	105	Network string `json:"network,omitempty"`
34afc569 Keenan Nemetz 16d ago	106	// Wait is how many seconds an idle lease may park before answering "nothing to
	107	// do". Zero answers at once, which is what a one-shot runner wants: it is here to
	108	// run the job it was started for, not to wait for the next one.
	109	Wait int `json:"wait,omitempty"`
b312e166 Keenan Nemetz 10d ago	110	// Record is the attempt's result, sent with finish. The runner no longer
	111	// publishes it to Git: this channel is where the server receives it, checks it
	112	// against the lease, and stores it.
	113	Record *runs.Record `json:"record,omitempty"`
5ece70a5 Keenan Nemetz 10d ago	114	// Log is one chunk of the attempt's log, sent while the job runs.
	115	Log *runnerLogChunk `json:"log,omitempty"`
	116	}
	117
	118	// runnerLogChunk is one piece of an attempt's log. Offset is where it starts in
	119	// the attempt's file, so the server can refuse a gap or a duplicate.
	120	type runnerLogChunk struct {
	121	Attempt int `json:"attempt"`
	122	Offset int64 `json:"offset"`
	123	Data []byte `json:"data"`
b44cbfaf Keenan Nemetz 17d ago	124	}
b44cbfaf Keenan Nemetz 17d ago	125
6e1fca51 Keenan Nemetz 6d ago	126	// runnerCredential is the private half of a run's key. It is delivered once, written
	127	// into the job's run directory, and never stored by the instance.
	128	type runnerCredential struct {
	129	PrivateKey string `json:"private_key"`
	130	Fingerprint string `json:"fingerprint"`
	131	}
	132
4ac42680 Keenan Nemetz 7d ago	133	// runnerSecret is one delivered value and the version it came from, so the attempt
	134	// record can say which value ran without carrying it.
	135	type runnerSecret struct {
	136	Version int `json:"version"`
	137	Value string `json:"value"`
	138	}
	139
	140	// runnerSecretIssue is a declared secret the repository could not supply. Reason is
	141	// empty when the reason is simply that nobody has stored that name.
	142	type runnerSecretIssue struct {
	143	Name string `json:"name"`
	144	Reason string `json:"reason,omitempty"`
	145	}
	146
b44cbfaf Keenan Nemetz 17d ago	147	type runnerStep struct {
	148	Name string `json:"name"`
	149	Run string `json:"run"`
	150	Timeout string `json:"timeout"`
	151	}
	152
c1929d2f Keenan Nemetz 5d ago	153	type runnerService struct {
	154	ID string `json:"id"`
	155	Image string `json:"image"`
	156	Env map[string]string `json:"env,omitempty"`
	157	}
	158
b44cbfaf Keenan Nemetz 17d ago	159	type runnerJob struct {
22528ac3 Keenan Nemetz 16d ago	160	ID string `json:"id"`
	161	Repo string `json:"repo"`
	162	SHA string `json:"sha"`
90fdaf39 Keenan Nemetz 16d ago	163	// Attempt is which try at this job this is. The record lives under it, so the
	164	// runner has to know the number it is reporting.
	165	Attempt int `json:"attempt"`
22528ac3 Keenan Nemetz 16d ago	166	// Ref, WorkflowSHA and StepsSHA are what the runner signs when the attempt ends:
22528ac3 Keenan Nemetz 16d ago	167	// the exact revision, the exact workflow file, and the digest of the steps that
90fdaf39 Keenan Nemetz 16d ago	168	// were pinned when the job was queued. Without them the record could only say
22528ac3 Keenan Nemetz 16d ago	169	// "some job", which is not a statement anybody can check later.
	170	Ref string `json:"ref"`
	171	WorkflowSHA string `json:"workflow_sha"`
	172	StepsSHA string `json:"steps_sha"`
	173	Branch string `json:"branch"`
	174	Workflow string `json:"workflow"`
	175	Job string `json:"job"`
	176	Steps []runnerStep `json:"steps"`
eb07ff56 Keenan Nemetz 8d ago	177	// Trigger and EventPayload are the pinned event that requested this run. The
	178	// runner writes the payload into a file and sets VOLE_EVENT, VOLE_EVENT_ACTION
	179	// and VOLE_EVENT_PATH; it is data, never an interpolated command.
	180	Trigger string `json:"trigger,omitempty"`
	181	TriggerAction string `json:"trigger_action,omitempty"`
	182	EventPayload string `json:"event_payload,omitempty"`
d7958024 Keenan Nemetz 6d ago	183	// Network is the isolation the job asked for with `network:`, or empty when the
	184	// workflow did not say. The runner applies it to the container it starts, so a job
	185	// that asks for none gets none even on a machine that could provide a bridge.
	186	Network string `json:"network,omitempty"`
6e1fca51 Keenan Nemetz 6d ago	187	// Bot is the account this job acts as, and Credential is the key it may do it with.
	188	// The private half is here and nowhere else: the queue holds the public half and
	189	// the account name, so a credential can be refused the moment the run ends.
	190	Bot string `json:"bot,omitempty"`
	191	Credential *runnerCredential `json:"credential,omitempty"`
	192	// CredentialError says the job asked to act as an account it cannot have. The
	193	// runner fails the job with this line rather than running it without the key,
	194	// because a step that cannot write is a broken job, not a degraded one.
	195	CredentialError string `json:"credential_error,omitempty"`
	196	// HostKey is where to find the instance when the job pushes, paired with the
	197	// credential above.
	198	HostKey string `json:"host_key,omitempty"`
e7aa0785 Keenan Nemetz 6d ago	199	// Timeout is the job's budget as a duration string, or empty when it has none.
e7aa0785 Keenan Nemetz 6d ago	200	Timeout string `json:"timeout,omitempty"`
70b70264 Keenan Nemetz 15d ago	201	// Image is a registry reference or "file:<name>". For the latter, ImageDefinition
	202	// is the pinned recipe and ImageSHA is its digest; the runner builds it locally.
	203	Image string `json:"image"`
	204	ImageFile string `json:"image_file,omitempty"`
	205	ImageDefinition string `json:"image_definition,omitempty"`
	206	ImageSHA string `json:"image_sha,omitempty"`
3b39a84c Keenan Nemetz 14d ago	207	// Env is the job's declared environment, pinned when the run was queued.
3b39a84c Keenan Nemetz 14d ago	208	Env map[string]string `json:"env,omitempty"`
c1929d2f Keenan Nemetz 5d ago	209	// Services are the containers the job asked to run beside it, pinned when the
	210	// run was queued. The runner gives the job and these one network for the attempt.
	211	Services []runnerService `json:"services,omitempty"`
4ac42680 Keenan Nemetz 7d ago	212	// Secrets are the values the job declared that this repository has, and Missing
	213	// are the names it declared that it does not. A missing name is not a failure:
	214	// the variable is left unset and the runner says so at the top of the log. The
	215	// values travel here and nowhere else — the queue row holds the names only.
	216	Secrets map[string]runnerSecret `json:"secrets,omitempty"`
	217	Missing []runnerSecretIssue `json:"missing_secrets,omitempty"`
213f4d6d Keenan Nemetz 17d ago	218	// Artifacts are the declared patterns the runner expands and uploads after the
	219	// steps succeed.
	220	Artifacts []string `json:"artifacts,omitempty"`
0f442b1c Keenan Nemetz 17d ago	221	// Tag is set for a tag-triggered run: the release being built. TagAnnotated says
	222	// whether that tag carries notes, which is what makes it a release.
	223	Tag string `json:"tag,omitempty"`
	224	TagAnnotated bool `json:"tag_annotated,omitempty"`
	225	// ReleaseArtifacts are the patterns that also publish to the tag's release.
	226	ReleaseArtifacts []string `json:"release_artifacts,omitempty"`
88e980b0 Keenan Nemetz 17d ago	227	// Pages is the site deploy this run performs, if any. The branch comes from
88e980b0 Keenan Nemetz 17d ago	228	// instance configuration, which is why it is in the payload rather than in the
c5902451 Keenan Nemetz 16d ago	229	// ci.
88e980b0 Keenan Nemetz 17d ago	230	Pages *runnerPages `json:"pages,omitempty"`
	231	}
	232
	233	type runnerPages struct {
	234	Path string `json:"path"`
	235	Ref string `json:"ref"`
b44cbfaf Keenan Nemetz 17d ago	236	}
	237
	238	type runnerReply struct {
	239	Op string `json:"op"`
	240	Error string `json:"error,omitempty"`
	241	Run *runnerJob `json:"run,omitempty"`
	242	OK bool `json:"ok"`
	243	Lease int `json:"lease,omitempty"`
62576fa8 Keenan Nemetz 14d ago	244	// Who am I: the account name this channel authenticated as. A runner answers
	245	// "ping" with this so it can say whose results it will be reporting, and so a
	246	// --repos filter can be checked against what the account may run.
	247	Name string `json:"name,omitempty"`
b238ac24 Keenan Nemetz 16d ago	248	// Repos are the repositories this account may run jobs for, sent with "ping" so
	249	// a runner started with --repos can find out at startup that its filter matches
	250	// nothing — a typo it would otherwise discover as a permanently idle runner —
	251	// instead of after the first build that never arrived.
	252	Repos []string `json:"repos,omitempty"`
5ece70a5 Keenan Nemetz 10d ago	253	// LogSize is how much of the log the server has after a log chunk.
5ece70a5 Keenan Nemetz 10d ago	254	LogSize int64 `json:"log_size,omitempty"`
0c14f078 Keenan Nemetz 6d ago	255	// Cancel names the account that asked this running job to stop. It rides the
	256	// heartbeat reply, so stopping a run needs no new traffic and a runner too old to
	257	// know the field ignores it. Empty means nobody has asked.
	258	Cancel string `json:"cancel,omitempty"`
b44cbfaf Keenan Nemetz 17d ago	259	}
	260
	261	// serveRunner answers one runner session until it disconnects.
62576fa8 Keenan Nemetz 14d ago	262	func (s *Server) serveRunner(ch xssh.Channel, user string) {
b44cbfaf Keenan Nemetz 17d ago	263	defer ch.Close()
	264	cfg := s.cfg
	265	if cfg.Runs == nil {
	266	_ = json.NewEncoder(ch).Encode(runnerReply{Op: "error", Error: "the run queue is not configured on this server"})
	267	return
	268	}
78cd3437 Keenan Nemetz 12d ago	269	dec := json.NewDecoder(&lineLimitReader{r: ch, limit: maxRunnerRequestBytes})
b44cbfaf Keenan Nemetz 17d ago	270	enc := json.NewEncoder(ch)
62576fa8 Keenan Nemetz 14d ago	271	owner := user
ac433531 Keenan Nemetz 16d ago	272	// Presence is one row per channel: a pool of three runners on one account is
	273	// three rows, and the row goes away when the channel does. Nothing else records
	274	// that a runner exists, so this is what the runners page shows.
	275	sessionID, err := id.New(time.Now().UTC())
	276	if err != nil {
	277	_ = enc.Encode(runnerReply{Op: "error", Error: err.Error()})
	278	return
	279	}
62576fa8 Keenan Nemetz 14d ago	280	if err := cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC()); err != nil {
ac433531 Keenan Nemetz 16d ago	281	cfg.Log.Warn("runner: cannot record the session", "user", user, "err", err)
	282	}
	283	defer cfg.Runs.DropSession(sessionID)
34afc569 Keenan Nemetz 16d ago	284	// repos is this session's eligibility list, reused for one poll window. Reading
	285	// every repository's rules on every lease request is O(repos) per idle runner, and
	286	// the answer changes only when an admin edits an allowlist — so it is refreshed on
	287	// the same five-second beat the roster is, and not once per request.
	288	var repos []string
	289	var reposAt time.Time
b44cbfaf Keenan Nemetz 17d ago	290	cfg.Log.Info("runner connected", "user", user)
	291
	292	for {
	293	var req runnerRequest
	294	if err := dec.Decode(&req); err != nil {
78cd3437 Keenan Nemetz 12d ago	295	if errors.Is(err, errRequestTooLarge) {
	296	_ = enc.Encode(runnerReply{Op: "error", Error: fmt.Sprintf(
	297	"a runner request must fit in %d bytes", maxRunnerRequestBytes)})
	298	cfg.Log.Warn("runner: refused an oversized request", "user", user)
	299	}
	300	return // disconnected, or refused
b44cbfaf Keenan Nemetz 17d ago	301	}
	302	switch req.Op {
	303	case "lease":
34afc569 Keenan Nemetz 16d ago	304	now := time.Now().UTC()
	305	// A runner may only run jobs for repositories whose runner allowlist it is
	306	// on: running a job executes that repository's code, which is a bigger
	307	// power than writing to it.
	308	if reposAt.IsZero() \|\| now.Sub(reposAt) > eligibleFor {
62576fa8 Keenan Nemetz 14d ago	309	list, err := s.runnableRepos(user)
34afc569 Keenan Nemetz 16d ago	310	if err != nil {
	311	_ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
	312	continue
	313	}
	314	repos, reposAt = list, now
b44cbfaf Keenan Nemetz 17d ago	315	}
b238ac24 Keenan Nemetz 16d ago	316	// --repos narrows the grant. The intersection is computed here rather than
	317	// trusted from the runner, so the worst a hostile client can do is ask for
	318	// nothing — never for a repository no allowlist gave it.
	319	want := narrow(repos, req.Repos)
34afc569 Keenan Nemetz 16d ago	320	// Take the wake-up before asking, so a run queued between the lease and the
	321	// wait closes a channel somebody is holding. Asking first and waiting second
	322	// loses exactly that race.
	323	wake := cfg.Runs.Work()
6c03e461 Keenan Nemetz 16d ago	324	lease := runs.LeaseRequest{
d7958024 Keenan Nemetz 6d ago	325	Owner: owner, Repos: want, Labels: req.Labels, Network: req.Network,
6c03e461 Keenan Nemetz 16d ago	326	TTL: cfg.Lease, MaxPerRepo: cfg.MaxJobsPerRepo,
6c03e461 Keenan Nemetz 16d ago	327	}
b238ac24 Keenan Nemetz 16d ago	328	run, err := cfg.Runs.Lease(lease, now)
b44cbfaf Keenan Nemetz 17d ago	329	if err != nil {
	330	_ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
	331	continue
	332	}
34afc569 Keenan Nemetz 16d ago	333	if run == nil && req.Wait > 0 {
	334	wait := time.Duration(req.Wait) * time.Second
	335	if wait > maxLeaseWait {
	336	wait = maxLeaseWait
	337	}
	338	select {
	339	case <-wake:
	340	case <-time.After(wait):
786d7ce1 Keenan Nemetz 2d ago	341	case <-s.done:
34afc569 Keenan Nemetz 16d ago	342	}
	343	// Ask again rather than assume: the wake-up may belong to a job another
	344	// runner took first, and a lease is the only thing that hands work out.
b238ac24 Keenan Nemetz 16d ago	345	run, err = cfg.Runs.Lease(lease, time.Now().UTC())
34afc569 Keenan Nemetz 16d ago	346	if err != nil {
	347	_ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
	348	continue
	349	}
	350	}
b44cbfaf Keenan Nemetz 17d ago	351	if run == nil {
ac433531 Keenan Nemetz 16d ago	352	// An idle runner is still connected: refresh its presence so the page can
6c03e461 Keenan Nemetz 16d ago	353	// say "waiting for work" instead of showing a runner that vanished. The
	354	// labels ride along, because a runner waiting for the wrong kind of work
	355	// is exactly what this row is for.
62576fa8 Keenan Nemetz 14d ago	356	_ = cfg.Runs.TouchSession(sessionID, user, req.Labels, time.Now().UTC())
b44cbfaf Keenan Nemetz 17d ago	357	_ = enc.Encode(runnerReply{Op: "lease", OK: true})
	358	continue
	359	}
4ac42680 Keenan Nemetz 7d ago	360	// The values are resolved here, at the moment the job is handed out, so
	361	// the queue never holds one and a rotation between queueing and leasing
	362	// delivers the value the job actually ran with.
	363	secrets, missing := s.leaseSecrets(run, user)
b44cbfaf Keenan Nemetz 17d ago	364	job := &runnerJob{
90fdaf39 Keenan Nemetz 16d ago	365	ID: run.ID, Repo: run.Repo, SHA: run.SHA, Attempt: run.Attempts, Branch: run.Branch,
22528ac3 Keenan Nemetz 16d ago	366	Ref: run.Ref, Workflow: run.Workflow, WorkflowSHA: run.WorkflowSHA,
22528ac3 Keenan Nemetz 16d ago	367	StepsSHA: run.StepsSHA, Job: run.Job, Artifacts: run.Artifacts,
eb07ff56 Keenan Nemetz 8d ago	368	Trigger: run.Trigger, TriggerAction: run.TriggerAction, EventPayload: run.EventPayload,
70b70264 Keenan Nemetz 15d ago	369	Image: run.Image, ImageFile: run.ImageFile, ImageDefinition: run.ImageDefinition,
70b70264 Keenan Nemetz 15d ago	370	ImageSHA: run.ImageSHA,
d7958024 Keenan Nemetz 6d ago	371	Network: run.Network,
e7aa0785 Keenan Nemetz 6d ago	372	Timeout: run.Timeout,
3b39a84c Keenan Nemetz 14d ago	373	Env: run.Env,
4ac42680 Keenan Nemetz 7d ago	374	Secrets: secrets, Missing: missing,
4ac42680 Keenan Nemetz 7d ago	375	Tag: run.Tag, TagAnnotated: run.TagAnnotated, ReleaseArtifacts: run.ReleaseArtifacts,
b44cbfaf Keenan Nemetz 17d ago	376	}
88e980b0 Keenan Nemetz 17d ago	377	if run.PagesPath != "" {
	378	ref := cfg.PagesRef
	379	if ref == "" {
	380	ref = "pages"
	381	}
	382	job.Pages = &runnerPages{Path: run.PagesPath, Ref: ref}
	383	}
b44cbfaf Keenan Nemetz 17d ago	384	for _, st := range run.Steps {
	385	job.Steps = append(job.Steps, runnerStep{Name: st.Name, Run: st.Run, Timeout: st.Timeout})
	386	}
c1929d2f Keenan Nemetz 5d ago	387	for _, svc := range run.Services {
	388	job.Services = append(job.Services, runnerService{ID: svc.ID, Image: svc.Image, Env: svc.Env})
	389	}
6e1fca51 Keenan Nemetz 6d ago	390	// A job that acts as a bot gets a key for this run only. The account and
	391	// the repository are pinned in the row, so the hook can refuse a write that
	392	// the account's role would not allow and the key stops working when the run
	393	// ends. A bot that cannot be used fails the job here rather than at its
	394	// first push, which is a much better place to learn it.
	395	if run.Bot != "" {
	396	cred, hostKey, reason := s.mintCredential(run, now)
	397	if reason != "" {
	398	job.CredentialError = reason
	399	} else {
	400	job.Bot, job.Credential, job.HostKey = run.Bot, cred, hostKey
	401	}
	402	}
b44cbfaf Keenan Nemetz 17d ago	403	cfg.Log.Info("runner leased a job", "user", user, "repo", run.Repo,
b44cbfaf Keenan Nemetz 17d ago	404	"job", run.Job, "sha", run.SHA[:8], "run", run.ID, "attempt", run.Attempts)
ac433531 Keenan Nemetz 16d ago	405	_ = cfg.Runs.SessionWorking(sessionID, run.Repo, run.ID, time.Now().UTC())
b44cbfaf Keenan Nemetz 17d ago	406	_ = enc.Encode(runnerReply{Op: "lease", OK: true, Run: job, Lease: int(cfg.Lease.Seconds())})
b44cbfaf Keenan Nemetz 17d ago	407
5ece70a5 Keenan Nemetz 10d ago	408	case "log":
	409	if req.Log == nil {
	410	_ = enc.Encode(runnerReply{Op: "log", Error: "a log request carries no chunk"})
	411	continue
	412	}
	413	run, err := cfg.Runs.Get(req.Run)
	414	if err != nil {
	415	_ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
	416	continue
	417	}
	418	if run == nil {
	419	_ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf("no run %s", req.Run)})
	420	continue
	421	}
	422	if req.Log.Attempt != run.Attempts {
	423	_ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf(
	424	"%s: the log is attempt %d but this runner holds attempt %d",
	425	req.Run, req.Log.Attempt, run.Attempts)})
	426	continue
	427	}
	428	held, err := cfg.Runs.Held(run.ID, owner)
	429	if err != nil {
	430	_ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
	431	continue
	432	}
	433	if !held {
	434	_ = enc.Encode(runnerReply{Op: "log", OK: false})
	435	continue
	436	}
	437	if cfg.Logs == nil {
	438	_ = enc.Encode(runnerReply{Op: "log", Error: "this instance has no log directory"})
	439	continue
	440	}
	441	size, err := cfg.Logs.Append(run.Repo, run.ID, req.Log.Attempt, req.Log.Offset, req.Log.Data)
	442	if err != nil {
	443	_ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
	444	continue
	445	}
	446	if ok, err := cfg.Runs.RecordLogSize(run.ID, owner, req.Log.Attempt, size, time.Now().UTC()); err != nil {
	447	_ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
	448	continue
	449	} else if !ok {
	450	_ = enc.Encode(runnerReply{Op: "log", OK: false})
	451	continue
	452	}
	453	_ = enc.Encode(runnerReply{Op: "log", OK: true, LogSize: size})
	454
b44cbfaf Keenan Nemetz 17d ago	455	case "heartbeat":
ac433531 Keenan Nemetz 16d ago	456	// Liveness only: the lease is what a heartbeat buys. Presence is refreshed
0c14f078 Keenan Nemetz 6d ago	457	// here too, so a long build does not make its runner look gone. A stop request
	458	// rides the same reply rather than a new message: the runner already asks this
	459	// question every few seconds, and a runner that does not know the field keeps
	460	// working exactly as before.
62576fa8 Keenan Nemetz 14d ago	461	_ = cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC())
0c14f078 Keenan Nemetz 6d ago	462	held, cancelBy, err := cfg.Runs.Heartbeat(req.Run, owner, cfg.Lease, time.Now().UTC())
b44cbfaf Keenan Nemetz 17d ago	463	if err != nil {
	464	_ = enc.Encode(runnerReply{Op: "heartbeat", Error: err.Error()})
	465	continue
	466	}
0c14f078 Keenan Nemetz 6d ago	467	_ = enc.Encode(runnerReply{Op: "heartbeat", OK: held, Cancel: cancelBy})
b44cbfaf Keenan Nemetz 17d ago	468
b44cbfaf Keenan Nemetz 17d ago	469	case "finish":
b312e166 Keenan Nemetz 10d ago	470	// The result arrives here, not on the metadata branch. The record is
	471	// checked against the lease and the job it claims to describe, then
	472	// stored with the run's status in one transaction.
ac433531 Keenan Nemetz 16d ago	473	_ = cfg.Runs.SessionIdle(sessionID, time.Now().UTC())
90fdaf39 Keenan Nemetz 16d ago	474	run, err := cfg.Runs.Get(req.Run)
22528ac3 Keenan Nemetz 16d ago	475	if err != nil {
	476	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
	477	continue
	478	}
90fdaf39 Keenan Nemetz 16d ago	479	if run == nil {
	480	_ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("no run %s", req.Run)})
	481	continue
22528ac3 Keenan Nemetz 16d ago	482	}
b312e166 Keenan Nemetz 10d ago	483	if req.Record == nil {
90fdaf39 Keenan Nemetz 16d ago	484	_ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
b312e166 Keenan Nemetz 10d ago	485	"%s: this runner did not send a record; upgrade the runner and the server together", req.Run)})
90fdaf39 Keenan Nemetz 16d ago	486	continue
b312e166 Keenan Nemetz 10d ago	487	}
b312e166 Keenan Nemetz 10d ago	488	if req.Record.Attempt != run.Attempts {
90fdaf39 Keenan Nemetz 16d ago	489	_ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
b312e166 Keenan Nemetz 10d ago	490	"%s: the record is attempt %d but this runner holds attempt %d",
b312e166 Keenan Nemetz 10d ago	491	req.Run, req.Record.Attempt, run.Attempts)})
90fdaf39 Keenan Nemetz 16d ago	492	continue
90fdaf39 Keenan Nemetz 16d ago	493	}
b312e166 Keenan Nemetz 10d ago	494	if req.Record.Signer != owner {
	495	_ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
	496	"%s: the record names %s as the reporter, but %s reported it",
	497	req.Run, req.Record.Signer, owner)})
379dad1e Keenan Nemetz 16d ago	498	continue
379dad1e Keenan Nemetz 16d ago	499	}
b312e166 Keenan Nemetz 10d ago	500	if err := checkRecordPointers(cfg.LFS, run.Repo, *req.Record); err != nil {
86fe5bb4 Keenan Nemetz 16d ago	501	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
86fe5bb4 Keenan Nemetz 16d ago	502	continue
b44cbfaf Keenan Nemetz 17d ago	503	}
5ece70a5 Keenan Nemetz 10d ago	504	if req.Record.LogSize == 0 {
	505	// A failed stream leaves a partial file, and an attempt that produced
	506	// no output leaves none. Either way the result names no log.
	507	if cfg.Logs != nil {
	508	if err := cfg.Logs.Remove(run.Repo, run.ID, req.Record.Attempt); err != nil {
	509	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
	510	continue
	511	}
	512	}
	513	} else {
	514	if cfg.Logs == nil {
	515	_ = enc.Encode(runnerReply{Op: "finish", Error: "this instance has no log directory"})
	516	continue
	517	}
	518	size, err := cfg.Logs.Size(run.Repo, run.ID, req.Record.Attempt)
	519	if err != nil {
	520	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
	521	continue
	522	}
	523	if size != req.Record.LogSize {
	524	_ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
	525	"%s: the log is %d bytes but the record says %d",
	526	req.Run, size, req.Record.LogSize)})
	527	continue
	528	}
	529	if err := cfg.Logs.Sync(run.Repo, run.ID, req.Record.Attempt); err != nil {
	530	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
	531	continue
	532	}
	533	}
b312e166 Keenan Nemetz 10d ago	534	ok, err := cfg.Runs.RecordAttempt(run.ID, owner, *req.Record, time.Now().UTC())
22528ac3 Keenan Nemetz 16d ago	535	if err != nil {
	536	_ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
	537	continue
	538	}
6e1fca51 Keenan Nemetz 6d ago	539	// The run is over, so the key it could act with is over too. The expiry would
	540	// catch it eventually; this is the part that does not wait.
	541	if err := cfg.Runs.DropRunCredentials(run.ID); err != nil {
	542	cfg.Log.Warn("runner: cannot drop the run credential", "run", run.ID, "err", err)
	543	}
b312e166 Keenan Nemetz 10d ago	544	cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", req.Record.Exit,
b312e166 Keenan Nemetz 10d ago	545	"signer", req.Record.Signer, "recorded", ok)
b44cbfaf Keenan Nemetz 17d ago	546	_ = enc.Encode(runnerReply{Op: "finish", OK: ok})
	547
	548	case "ping":
62576fa8 Keenan Nemetz 14d ago	549	reply := runnerReply{Op: "ping", OK: true, Name: user}
b238ac24 Keenan Nemetz 16d ago	550	// What this account may run, so a runner started with --repos can say at
b238ac24 Keenan Nemetz 16d ago	551	// startup that its filter matches nothing rather than idling in silence.
62576fa8 Keenan Nemetz 14d ago	552	if granted, gerr := s.runnableRepos(user); gerr == nil {
b238ac24 Keenan Nemetz 16d ago	553	reply.Repos = granted
	554	} else {
	555	cfg.Log.Warn("runner: cannot list runnable repositories", "user", user, "err", gerr)
	556	}
22528ac3 Keenan Nemetz 16d ago	557	_ = enc.Encode(reply)
b44cbfaf Keenan Nemetz 17d ago	558
	559	default:
	560	_ = enc.Encode(runnerReply{Op: req.Op, Error: "unknown op"})
	561	}
	562	}
	563	}
	564
b312e166 Keenan Nemetz 10d ago	565	// checkRecordPointers refuses a record whose log or artifacts are not actually in
	566	// the object store. A push used to be checked for pointer hygiene at the boundary;
	567	// the finish channel is the boundary now.
	568	func checkRecordPointers(objects *lfs.Store, repoName string, rec runs.Record) error {
	569	for _, a := range rec.Artifacts {
	570	if err := checkPointer(objects, repoName, lfs.Pointer{OID: a.OID, Size: a.Size}); err != nil {
	571	return fmt.Errorf("artifact %s: %w", a.Name, err)
	572	}
	573	}
	574	return nil
	575	}
	576
	577	func checkPointer(objects *lfs.Store, repoName string, p lfs.Pointer) error {
	578	if objects == nil {
	579	return fmt.Errorf("this instance has no object store for run outputs")
	580	}
	581	size, err := objects.Stat(repoName, p.OID)
	582	if err != nil {
	583	return fmt.Errorf("object %s is missing: %w", p.OID, err)
	584	}
	585	if size != p.Size {
	586	return fmt.Errorf("object %s is %d bytes, pointer says %d", p.OID, size, p.Size)
	587	}
	588	return nil
	589	}
	590
b44cbfaf Keenan Nemetz 17d ago	591	// runnableRepos lists the repositories this account may execute jobs for.
62576fa8 Keenan Nemetz 14d ago	592	func (s *Server) runnableRepos(name string) ([]string, error) {
b44cbfaf Keenan Nemetz 17d ago	593	repos, err := s.cfg.Repos.List()
	594	if err != nil {
	595	return nil, err
	596	}
	597	var out []string
	598	for _, m := range repos {
	599	a, err := m.ACL()
	600	if err != nil {
	601	continue // a repository without readable rules grants nothing
	602	}
852c067d Keenan Nemetz 16d ago	603	// Whose results count is the same list as whose job may be leased: running a
	604	// job executes that repository's code, and a result from an account the
	605	// repository will not count is only worth refusing here rather than after
	606	// the build.
62576fa8 Keenan Nemetz 14d ago	607	if a.CanReport(name) {
b44cbfaf Keenan Nemetz 17d ago	608	out = append(out, m.Name)
	609	}
	610	}
	611	return out, nil
	612	}
b238ac24 Keenan Nemetz 16d ago	613
	614	// narrow keeps only the repositories a runner asked for, and is the whole of what
	615	// --repos is allowed to do. The granted list is what the account's runner allowlists
	616	// produce; the asked list comes from the client. Intersecting them means the worst a
	617	// client can do is run less than it was allowed to — the alternative, honouring the
	618	// client's list, would make an SSH flag an access-control decision.
	619	//
	620	// An empty asked list means "no narrowing" rather than "nothing": that is what a runner
	621	// without --repos sends, and it keeps such a runner following allowlist edits instead
	622	// of freezing its list at connect time. Asked names that were never granted simply
	623	// disappear, which the runner reports on startup because it can see both lists.
	624	func narrow(granted, asked []string) []string {
	625	if len(asked) == 0 {
	626	return granted
	627	}
	628	want := make(map[string]bool, len(asked))
	629	for _, r := range asked {
	630	want[r] = true
	631	}
	632	out := make([]string, 0, len(granted))
	633	for _, r := range granted {
	634	if want[r] {
	635	out = append(out, r)
	636	}
	637	}
	638	return out
	639	}