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.
Store CI run records in the database, not the repository
b312e1660ac3d570fc35350a22b3514c56ab466f · Keenan Nemetz · 11d ago · parent db5e4382
Store CI run records in the database, not the repository A finished attempt was a set of files on the vole branch: a record, step ranges, and artifact pointers, pushed as a commit. At 959 files, CI records were 849 of them, and every human metadata write had to carry them. The runner now sends the record with its finish message over the SSH channel that holds the lease. The server checks the account, the job, and the log and artifact pointers, then writes the attempt, steps, artifacts, attestation, and run result in one transaction in data/runs.db. A tagged build still pushes its release pointers to the branch after the record is stored. Pre-receive refuses a new runs/ file. vole admin index rebuild imports records still on an old branch; records written since are never deleted by a rebuild. The run page reads the stored attempt, and runs.db is now a source that must be backed up. This stops new CI history from entering Git. Existing commits keep the records they already have; removing those would require rewriting the protected vole branch, which this change does not do.
server/ssh/runner.go +72−100
| @@ -9,7 +9,7 @@ import ( | |||
| 9 | 9 | ||
| 10 | 10 | "github.com/rodent-software/vole/ci/runs" | |
| 11 | 11 | "github.com/rodent-software/vole/core/id" | |
| 12 | "github.com/rodent-software/vole/metadata" | ||
| 12 | "github.com/rodent-software/vole/core/lfs" | ||
| 13 | 13 | xssh "golang.org/x/crypto/ssh" | |
| 14 | 14 | ) | |
| 15 | 15 | ||
| @@ -22,19 +22,23 @@ import ( | |||
| 22 | 22 | // | |
| 23 | 23 | // {"op":"lease"} -> {"run":{...}} or {} | |
| 24 | 24 | // {"op":"heartbeat","run":"<id>"} -> {"ok":true|false} | |
| 25 | // {"op":"finish","run":"<id>"} -> {"ok":true|false} | ||
| 25 | // {"op":"finish","run":"<id>","record":{...}} -> {"ok":true|false} | ||
| 26 | 26 | // | |
| 27 | // A finish carries no result: the result is on the branch already. The runner | ||
| 28 | // uploads its log and artifacts, commits the attempt's record under | ||
| 29 | // `runs/<run id>/<attempt>/` together with any release pointers the build published, | ||
| 30 | // pushes that once, and only then says it is done. `pre-receive` is what accepts or | ||
| 31 | // refuses it, which is where the runner allowlist and the signature are checked; this | ||
| 32 | // channel can only close the lease and mirror what the push already established. | ||
| 33 | // Nothing here writes a result, so there is nothing here for the instance to vouch | ||
| 34 | // for. | ||
| 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. | ||
| 35 | 37 | // | |
| 36 | 38 | // "ok":false means the lease is gone — another runner has the job — and the runner | |
| 37 | // must stop rather than report a result nobody will record. | ||
| 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. | ||
| 38 | 42 | ||
| 39 | 43 | const ( | |
| 40 | 44 | // eligibleFor is how long a session reuses its repository list. Access rules live | |
| @@ -44,11 +48,11 @@ const ( | |||
| 44 | 48 | // maxLeaseWait caps what a runner may ask for. The wait is a hint — the lease | |
| 45 | 49 | // remains the database's decision — so all a long wait holds is one SSH channel. | |
| 46 | 50 | maxLeaseWait = 60 * time.Second | |
| 47 | // maxRunnerRequestBytes bounds one request line. A runner request is a few | ||
| 48 | // hundred bytes of JSON; the field a hostile client would grow is --repos, and | ||
| 49 | // each name in it is looked up against the instance's repositories. The bound | ||
| 50 | // is per line, so a long-lived runner is not limited in total. | ||
| 51 | maxRunnerRequestBytes = 1 << 20 | ||
| 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 | ||
| 52 | 56 | ) | |
| 53 | 57 | ||
| 54 | 58 | // errRequestTooLarge ends a runner session whose request line outgrew the bound. | |
| @@ -97,6 +101,10 @@ type runnerRequest struct { | |||
| 97 | 101 | // do". Zero answers at once, which is what a one-shot runner wants: it is here to | |
| 98 | 102 | // run the job it was started for, not to wait for the next one. | |
| 99 | 103 | Wait int `json:"wait,omitempty"` | |
| 104 | // Record is the attempt's result, sent with finish. The runner no longer | ||
| 105 | // publishes it to Git: this channel is where the server receives it, checks it | ||
| 106 | // against the lease, and stores it. | ||
| 107 | Record *runs.Record `json:"record,omitempty"` | ||
| 100 | 108 | } | |
| 101 | 109 | ||
| 102 | 110 | type runnerStep struct { | |
| @@ -302,12 +310,9 @@ func (s *Server) serveRunner(ch xssh.Channel, user string) { | |||
| 302 | 310 | _ = enc.Encode(runnerReply{Op: "heartbeat", OK: ok}) | |
| 303 | 311 | ||
| 304 | 312 | case "finish": | |
| 305 | // Nothing is created here. The runner's record is already on the branch, | ||
| 306 | // checked by `pre-receive`; this closes the lease and mirrors the record | ||
| 307 | // into the queue so a page or a gate needs no tree walk. | ||
| 308 | // | ||
| 309 | // The session goes idle either way: whether the finish is accepted or the | ||
| 310 | // lease was already taken away, this runner is no longer building anything. | ||
| 313 | // The result arrives here, not on the metadata branch. The record is | ||
| 314 | // checked against the lease and the job it claims to describe, then | ||
| 315 | // stored with the run's status in one transaction. | ||
| 311 | 316 | _ = cfg.Runs.SessionIdle(sessionID, time.Now().UTC()) | |
| 312 | 317 | run, err := cfg.Runs.Get(req.Run) | |
| 313 | 318 | if err != nil { | |
| @@ -318,98 +323,34 @@ func (s *Server) serveRunner(ch xssh.Channel, user string) { | |||
| 318 | 323 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("no run %s", req.Run)}) | |
| 319 | 324 | continue | |
| 320 | 325 | } | |
| 321 | now := time.Now().UTC() | ||
| 322 | // The queue may already have given this job away: a runner whose lease was | ||
| 323 | // taken over is finishing an attempt somebody else owns. | ||
| 324 | held, err := cfg.Runs.Held(run.ID, owner) | ||
| 325 | if err != nil { | ||
| 326 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | ||
| 327 | continue | ||
| 328 | } | ||
| 329 | if !held { | ||
| 330 | _ = enc.Encode(runnerReply{Op: "finish", OK: false}) | ||
| 331 | continue | ||
| 332 | } | ||
| 333 | m, err := cfg.Repos.Open(run.Repo) | ||
| 334 | if err != nil { | ||
| 335 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("cannot open %s: %v", run.Repo, err)}) | ||
| 336 | continue | ||
| 337 | } | ||
| 338 | rec, err := runs.ReadRecord(m.Git(), metadata.Ref, run.ID) | ||
| 339 | if err != nil { | ||
| 340 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("cannot read the record for %s: %v", run.ID, err)}) | ||
| 341 | continue | ||
| 342 | } | ||
| 343 | switch { | ||
| 344 | case rec == nil: | ||
| 345 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | ||
| 346 | "no record for run %s on %s: a runner reports a result by pushing %s, and nothing is recorded without one", | ||
| 347 | run.ID, metadata.Ref, runs.RecordPath(run.ID, run.Attempts))}) | ||
| 348 | continue | ||
| 349 | case rec.Attempt != run.Attempts: | ||
| 326 | if req.Record == nil { | ||
| 350 | 327 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | |
| 351 | "the record for %s is attempt %d but this runner holds attempt %d", | ||
| 352 | run.ID, rec.Attempt, run.Attempts)}) | ||
| 328 | "%s: this runner did not send a record; upgrade the runner and the server together", req.Run)}) | ||
| 353 | 329 | continue | |
| 354 | case rec.Signer != user: | ||
| 330 | } | ||
| 331 | if req.Record.Attempt != run.Attempts { | ||
| 355 | 332 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | |
| 356 | "the record for %s was reported by another account; %s cannot finish it", | ||
| 357 | run.ID, user)}) | ||
| 333 | "%s: the record is attempt %d but this runner holds attempt %d", | ||
| 334 | req.Run, req.Record.Attempt, run.Attempts)}) | ||
| 358 | 335 | continue | |
| 359 | 336 | } | |
| 360 | if len(rec.Artifacts) > 0 { | ||
| 361 | ok, err := cfg.Runs.RecordArtifacts(run.ID, owner, rec.Artifacts) | ||
| 362 | if err != nil { | ||
| 363 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | ||
| 364 | continue | ||
| 365 | } | ||
| 366 | if !ok { | ||
| 367 | _ = enc.Encode(runnerReply{Op: "finish", OK: false}) | ||
| 368 | continue | ||
| 369 | } | ||
| 370 | } | ||
| 371 | if rec.Log != nil { | ||
| 372 | recorded, err := cfg.Runs.RecordLog(run.ID, owner, rec.Log.OID, rec.Log.Size, now) | ||
| 373 | if err != nil { | ||
| 374 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | ||
| 375 | continue | ||
| 376 | } | ||
| 377 | if !recorded { | ||
| 378 | _ = enc.Encode(runnerReply{Op: "finish", OK: false}) | ||
| 379 | continue | ||
| 380 | } | ||
| 381 | } | ||
| 382 | // The machine and container environment are claims from the record, mirrored | ||
| 383 | // under the same rule as the log and the artifacts: written only while this | ||
| 384 | // runner still holds the lease, so a runner that was overtaken leaves no claim | ||
| 385 | // behind. | ||
| 386 | if mirrored, err := cfg.Runs.RecordEnvironment(run.ID, owner, rec.OS, rec.Arch, | ||
| 387 | rec.Image, rec.ImageFile, rec.ImageSHA, rec.Runtime); err != nil { | ||
| 388 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | ||
| 389 | continue | ||
| 390 | } else if !mirrored { | ||
| 391 | _ = enc.Encode(runnerReply{Op: "finish", OK: false}) | ||
| 337 | if req.Record.Signer != owner { | ||
| 338 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | ||
| 339 | "%s: the record names %s as the reporter, but %s reported it", | ||
| 340 | req.Run, req.Record.Signer, owner)}) | ||
| 392 | 341 | continue | |
| 393 | 342 | } | |
| 394 | attested, err := cfg.Runs.RecordAttestation(run.ID, owner, runs.Attestation{ | ||
| 395 | Signer: rec.Signer, | ||
| 396 | CommitID: rec.Commit, | ||
| 397 | }, now) | ||
| 398 | if err != nil { | ||
| 343 | if err := checkRecordPointers(cfg.LFS, run.Repo, *req.Record); err != nil { | ||
| 399 | 344 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 400 | 345 | continue | |
| 401 | 346 | } | |
| 402 | if !attested { | ||
| 403 | _ = enc.Encode(runnerReply{Op: "finish", OK: false}) | ||
| 404 | continue | ||
| 405 | } | ||
| 406 | ok, err := cfg.Runs.Finish(run.ID, owner, rec.Exit, now) | ||
| 347 | ok, err := cfg.Runs.RecordAttempt(run.ID, owner, *req.Record, time.Now().UTC()) | ||
| 407 | 348 | if err != nil { | |
| 408 | 349 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 409 | 350 | continue | |
| 410 | 351 | } | |
| 411 | cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", rec.Exit, | ||
| 412 | "signer", rec.Signer, "commit", rec.Commit, "recorded", ok) | ||
| 352 | cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", req.Record.Exit, | ||
| 353 | "signer", req.Record.Signer, "recorded", ok) | ||
| 413 | 354 | _ = enc.Encode(runnerReply{Op: "finish", OK: ok}) | |
| 414 | 355 | ||
| 415 | 356 | case "ping": | |
| @@ -429,6 +370,37 @@ func (s *Server) serveRunner(ch xssh.Channel, user string) { | |||
| 429 | 370 | } | |
| 430 | 371 | } | |
| 431 | 372 | ||
| 373 | // checkRecordPointers refuses a record whose log or artifacts are not actually in | ||
| 374 | // the object store. A push used to be checked for pointer hygiene at the boundary; | ||
| 375 | // the finish channel is the boundary now. | ||
| 376 | func checkRecordPointers(objects *lfs.Store, repoName string, rec runs.Record) error { | ||
| 377 | if rec.Log != nil { | ||
| 378 | if err := checkPointer(objects, repoName, *rec.Log); err != nil { | ||
| 379 | return fmt.Errorf("log: %w", err) | ||
| 380 | } | ||
| 381 | } | ||
| 382 | for _, a := range rec.Artifacts { | ||
| 383 | if err := checkPointer(objects, repoName, lfs.Pointer{OID: a.OID, Size: a.Size}); err != nil { | ||
| 384 | return fmt.Errorf("artifact %s: %w", a.Name, err) | ||
| 385 | } | ||
| 386 | } | ||
| 387 | return nil | ||
| 388 | } | ||
| 389 | |||
| 390 | func checkPointer(objects *lfs.Store, repoName string, p lfs.Pointer) error { | ||
| 391 | if objects == nil { | ||
| 392 | return fmt.Errorf("this instance has no object store for run outputs") | ||
| 393 | } | ||
| 394 | size, err := objects.Stat(repoName, p.OID) | ||
| 395 | if err != nil { | ||
| 396 | return fmt.Errorf("object %s is missing: %w", p.OID, err) | ||
| 397 | } | ||
| 398 | if size != p.Size { | ||
| 399 | return fmt.Errorf("object %s is %d bytes, pointer says %d", p.OID, size, p.Size) | ||
| 400 | } | ||
| 401 | return nil | ||
| 402 | } | ||
| 403 | |||
| 432 | 404 | // runnableRepos lists the repositories this account may execute jobs for. | |
| 433 | 405 | func (s *Server) runnableRepos(name string) ([]string, error) { | |
| 434 | 406 | repos, err := s.cfg.Repos.List() | |