Sign in

Search issue titles and bodies, change requests and comments. Private repositories you cannot read are not included.

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.

Phase 4: the runner uploads the log, so the server cannot write it

86fe5bb49ddeeb89a5d58459f27a86a83f82481a · Keenan Nemetz · 17d ago · parent 9cc2f087

Phase 4: the runner uploads the log, so the server cannot write it

Streaming made the server the author of bytes the runner produced. Heartbeats
carried output chunks; the server concatenated them into a SQLite column and, at
finish, hashed the result itself with `LFS.PutBytes`. So the archived object's id was
the server's, not the writer's — the one output in the system that worked that way,
since artifacts are uploaded by the runner through put-object and pages/releases are
pushed by it. Nothing depended on it, which is why it was harmless; a signed run
record could never have bound it, which is why it had to go.

Streaming is dropped. At finish the runner uploads its log through the same
git-lfs-transfer path artifacts use, gets the object id the store computed from the
content, and names it in the finish message. The server's rule is the artifact rule:
the store is the authority on whether the bytes exist, and a claimed size that is not
the object's size is refused. Heartbeats now carry nothing but the lease.

A log is per attempt, not per run. A job that kills a runner is requeued, so one run
can have several logs, and the interesting one is usually the attempt that died:
`run_logs(run_id, attempt, oid, size)` replaces the single column, `AttemptLogs`
reads them oldest first, `/runs/<id>/log` serves the newest and `?attempt=N` picks
one, and `vole gc` treats every attempt's object as a root. Rows written before this
change still have their `log_oid`, which reads keep honouring; `Store.Log` and
`SetLogOID` are gone with the code that used them.

The cost, stated plainly: there is no live tail any more, and an attempt whose runner
was killed leaves no log at all — the attempt that hung is exactly the log somebody
wants. That is the trade: a log that can be shown to be the runner's is worth more
than one that arrives early. It also removes an unbounded text column from the queue,
which was growing into the WAL and every backup until a run finished.

Verified: TestCheckRunLogTrustsTheStore pins the new rule at the boundary (stored log
accepted, size mismatch refused, object absent refused, non-oid refused, no store
refused); TestLogsAreRecordedPerAttempt shows two attempts recorded separately,
`Get` returning the newest, both objects named for the sweep, and a finished run
refusing a log; TestHeartbeatAndFinishNeedTheLease now proves a stranger can neither
heartbeat, nor finish, nor record a log, and that a finished run carries no log text;
and the end-to-end test that runs the real binary now asserts the served log hashes
to the object id in the run row and that the queue kept no copy of it. scripts/ci.sh
green (gofmt, vet, build, markdown links, go test -race ./...).

internal/sshserver/runner.go +56−28

@@ -22,18 +22,29 @@ const maxRunArtifacts = 64
22 22 // which repositories it may run jobs for. One JSON object per line, one reply per
23 23 // request, on a single channel.
24 24 //
25 // {"op":"lease"} -> {"run":{...}} or {}
26 // {"op":"heartbeat","run":"<id>","log":"..."} -> {"ok":true|false}
27 // {"op":"finish","run":"<id>","exit":0} -> {"ok":true|false}
25 // {"op":"lease"} -> {"run":{...}} or {}
26 // {"op":"heartbeat","run":"<id>"} -> {"ok":true|false}
27 // {"op":"finish","run":"<id>","exit":0,
28 // "log_oid":"<sha256>","log_size":1234} -> {"ok":true|false}
28 29 //
29 30 // "ok":false means the lease is gone — another runner has the job — and the runner
30 31 // must stop rather than report a result nobody will record.
32 //
33 // A log is not streamed. The runner uploads its own bytes through the same
34 // git-lfs-transfer put-object path it uses for artifacts, and names the object here:
35 // the id is computed from the content by the party that produced it, so the
36 // archived log is something that party can attest to rather than a record the
37 // server assembled.
31 38
32 39 type runnerRequest struct {
33 40 Op string `json:"op"`
34 41 Run string `json:"run"`
35 Log string `json:"log"`
36 42 Exit int `json:"exit"`
43 // LogOID names the log object the runner uploaded for this attempt, and
44 // LogSize is how many bytes it holds. Both are absent when a log could not be
45 // stored: a result is worth recording even when its output is not.
46 LogOID string `json:"log_oid,omitempty"`
47 LogSize int64 `json:"log_size,omitempty"`
37 48 // Artifacts arrive with the finish, not as a separate call: a job's outputs
38 49 // and its result are one fact, so there is no window in which a successful run
39 50 // has no artifacts or a failed one has some.
@@ -145,7 +156,8 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
145 156 _ = enc.Encode(runnerReply{Op: "lease", OK: true, Run: job, Lease: int(cfg.Lease.Seconds())})
146 157
147 158 case "heartbeat":
148 ok, err := cfg.Runs.Heartbeat(req.Run, owner, req.Log, cfg.Lease, time.Now().UTC())
159 // Liveness only: the lease is what a heartbeat buys.
160 ok, err := cfg.Runs.Heartbeat(req.Run, owner, cfg.Lease, time.Now().UTC())
149 161 if err != nil {
150 162 _ = enc.Encode(runnerReply{Op: "heartbeat", Error: err.Error()})
151 163 continue
@@ -173,34 +185,31 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
173 185 continue
174 186 }
175 187 }
176 // A finished log never changes again, so it becomes a content-addressed
177 // object and the queue row stops carrying text. The object is written
178 // first: if this runner has lost the lease, the row is left alone and
179 // the object is simply swept later.
180 ok, err := false, error(nil)
181 if cfg.LFS != nil {
182 text := ""
183 if prev, lerr := cfg.Runs.Log(req.Run); lerr == nil {
184 text = prev + req.Log
185 }
186 if text != "" {
187 oid, perr := cfg.LFS.PutBytes([]byte(text))
188 if perr != nil {
189 cfg.Log.Warn("run log store", "run", req.Run, "err", perr)
190 } else {
191 ok, err = cfg.Runs.FinishAndArchive(req.Run, owner, req.Exit, oid, time.Now().UTC())
192 }
193 }
194 }
195 if cfg.LFS == nil || err != nil || !ok {
196 if err == nil && !ok {
197 // Either no store, or an empty log: fall back to the plain finish.
198 ok, err = cfg.Runs.Finish(req.Run, owner, req.Exit, req.Log, time.Now().UTC())
188 // The log is the runner's own bytes, uploaded to the content-addressed
189 // store before this message: the store is the authority on whether they
190 // exist, exactly as for artifacts, and the object id was computed by the
191 // party that wrote them. The server never assembles a log, which is what
192 // makes the archived bytes something a runner can attest to.
193 if req.LogOID != "" {
194 if err := s.checkRunLog(req.LogOID, req.LogSize); err != nil {
195 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
196 continue
199 197 }
198 recorded, err := cfg.Runs.RecordLog(req.Run, owner, req.LogOID, req.LogSize, time.Now().UTC())
200 199 if err != nil {
201 200 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
202 201 continue
203 202 }
203 if !recorded {
204 // The lease moved: report nothing, like a lost finish.
205 _ = enc.Encode(runnerReply{Op: "finish", OK: false})
206 continue
207 }
208 }
209 ok, err := cfg.Runs.Finish(req.Run, owner, req.Exit, time.Now().UTC())
210 if err != nil {
211 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
212 continue
204 213 }
205 214 cfg.Log.Info("run finished", "user", user, "run", req.Run, "exit", req.Exit, "recorded", ok)
206 215 _ = enc.Encode(runnerReply{Op: "finish", OK: ok})
@@ -217,6 +226,25 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
217 226 // checkArtifacts validates what a runner claims it uploaded. The store is the
218 227 // authority on whether the bytes exist; this only refuses claims that are not
219 228 // downloadable at all.
229 // checkRunLog verifies a log claim the way checkArtifacts verifies an artifact
230 // claim: the bytes must be in the store, and the declared size must be theirs.
231 func (s *Server) checkRunLog(oid string, size int64) error {
232 if !lfs.ValidOID(oid) {
233 return fmt.Errorf("log: %q is not a sha256 object id", oid)
234 }
235 if s.cfg.LFS == nil {
236 return fmt.Errorf("this instance has no object store, so a log cannot be archived")
237 }
238 n, err := s.cfg.LFS.Stat(oid)
239 if err != nil {
240 return fmt.Errorf("log names object %s, which is not in the store: upload it before finishing", oid[:12])
241 }
242 if size != n {
243 return fmt.Errorf("log says %d bytes but the stored object is %d", size, n)
244 }
245 return nil
246 }
247
220 248 func (s *Server) checkArtifacts(in []runnerArtifact) ([]runs.Artifact, error) {
221 249 if len(in) > maxRunArtifacts {
222 250 return nil, fmt.Errorf("%d artifacts, limit %d", len(in), maxRunArtifacts)