vole

Clone

Read only, no credentials:

https://rodent.software/vole.git

Write over SSH with a registered key:

git clone ssh://git@rodent.software:2222/vole.git

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

Renew the lease at every step boundary

539d6e6cd19ab148c18467d3b9dc9746f30d87b7 · Keenan Nemetz · 1d ago · parent c980fe53

Renew the lease at every step boundary

A step heartbeats while it runs, but its ticker starts with the step, so a step
that finishes inside one interval never heartbeats at all. An attempt whose
steps add up to more than the lease therefore expired between two of them, and
the next heartbeat — the first tick of the long step at the end — arrived to
find the queue had already given the job away. Twelve short steps in front of
`go test` is exactly that shape: it killed attempts at 61s, 65s and 77s with no
result, five times per run, and the retry re-ran every step from the start.

Send one heartbeat before each step, which is the only place a step shorter than
the interval can renew. A lease that is really gone still stops the attempt, and
the command ticker keeps covering a step that runs longer than one interval.

Verified on a runner at the default lease_seconds = 60: a nine-step workflow
that died four times at the boundary ran to completion inside one lease, and
failed only on the ci/cache test that #236 covers.

ci/agent/heartbeat_test.go +84

@@ -0,0 +1,84 @@
	1	package agent
	2
	3	import (
	4	"bufio"
	5	"encoding/json"
	6	"io"
	7	"testing"
	8	)
	9
	10	// captureSession is a session with no ssh under it that records the requests it is
	11	// given, so a test can say what the runner asked for and not only what it did with
	12	// the answer. One reply is written per request.
	13	func captureSession(t testing.T, reqs chan<- runnerRequest, replies ...runnerReply) runnerSession {
	14	t.Helper()
	15	reqR, reqW := io.Pipe()
	16	repR, repW := io.Pipe()
	17	go func() {
	18	defer repW.Close()
	19	dec := json.NewDecoder(reqR)
	20	enc := json.NewEncoder(repW)
	21	for _, reply := range replies {
	22	var req runnerRequest
	23	if err := dec.Decode(&req); err != nil {
	24	return
	25	}
	26	select {
	27	case reqs <- req:
	28	default:
	29	}
	30	if err := enc.Encode(reply); err != nil {
	31	return
	32	}
	33	}
	34	}()
	35	t.Cleanup(func() { _ = reqW.Close() })
	36	return &runnerSession{
	37	in: bufio.NewWriter(reqW),
	38	out: bufio.NewReader(repR),
	39	enc: json.NewEncoder(reqW),
	40	dec: json.NewDecoder(repR),
	41	}
	42	}
	43
	44	// The renewal between steps has to be the request the server expects, carrying the
	45	// run it belongs to: the whole point of the step boundary is that no command is
	46	// running to send it.
	47	func TestHeartbeatBetweenStepsNamesTheRun(t *testing.T) {
	48	t.Parallel()
	49	for _, held := range []bool{true, false} {
	50	reqs := make(chan runnerRequest, 1)
	51	s := captureSession(t, reqs, runnerReply{Op: "heartbeat", OK: held})
	52	if got := s.heartbeat("run-1"); got != held {
	53	t.Fatalf("heartbeat = %v, want %v", got, held)
	54	}
	55	select {
	56	case req := <-reqs:
	57	if req.Op != "heartbeat" \|\| req.Run != "run-1" {
	58	t.Fatalf("heartbeat sent %+v; want op=heartbeat run=run-1", req)
	59	}
	60	default:
	61	t.Fatal("heartbeat sent no request")
	62	}
	63	}
	64	}
	65
	66	// A heartbeat that cannot be sent is not a lease: the attempt must stop rather than
	67	// run a step the queue may already have given to another runner.
	68	func TestHeartbeatWithoutAConnectionIsNotHeld(t *testing.T) {
	69	t.Parallel()
	70	reqR, reqW := io.Pipe()
	71	repR, repW := io.Pipe()
	72	_ = reqR.Close()
	73	_ = repW.Close()
	74	t.Cleanup(func() { _ = reqW.Close(); _ = repR.Close() })
	75	s := &runnerSession{
	76	in: bufio.NewWriter(reqW),
	77	out: bufio.NewReader(repR),
	78	enc: json.NewEncoder(reqW),
	79	dec: json.NewDecoder(repR),
	80	}
	81	if s.heartbeat("run-1") {
	82	t.Fatal("a heartbeat on a dead connection reported the lease as held")
	83	}
	84	}