detached 8784b11

fix(guest): escalate cancellation SIGINT/SIGTERM/SIGKILL, stream through frame writers, forward init signals; cancel channel in Drive

commit
8784b11a7a400942fc6c0c2dda70628c27f9cf57
author
brice · · 2026-10-12 16:00 UTC
parent
af04fdc
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
16 files · +682 -145
Unified
UnifiedSplit
A buf.gen.yaml +5 -0
@@ -0,0 +1,5 @@
1+version: v2
2+plugins:
3+ - local: [go, tool, protoc-gen-go]
4+ out: ../..
5+ opt: module=bnema.dev/ermite-runner
@@ -0,0 +1,5 @@
1+version: v2
2+plugins:
3+ - local: [go, tool, protoc-gen-go]
4+ out: ../..
5+ opt: module=bnema.dev/ermite-runner
A buf.yaml +12 -0
@@ -0,0 +1,12 @@
1+version: v2
2+modules:
3+ - path: internal/execstream
4+lint:
5+ use:
6+ - STANDARD
7+ except:
8+ # The proto is internal to this module and lives next to its Go package.
9+ - PACKAGE_DIRECTORY_MATCH
10+breaking:
11+ use:
12+ - FILE
@@ -0,0 +1,12 @@
1+version: v2
2+modules:
3+ - path: internal/execstream
4+lint:
5+ use:
6+ - STANDARD
7+ except:
8+ # The proto is internal to this module and lives next to its Go package.
9+ - PACKAGE_DIRECTORY_MATCH
10+breaking:
11+ use:
12+ - FILE
M go.mod +76 -6
@@ -12,25 +12,95 @@ require (
1212 )
1313
1414 require (
15+ buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
16+ buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
17+ buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
18+ buf.build/gen/go/bufbuild/registry/connectrpc/go v1.20.0-20260831211851-b4432e12a6e7.1 // indirect
19+ buf.build/gen/go/bufbuild/registry/protocolbuffers/go v1.36.12-20260831211851-b4432e12a6e7.2 // indirect
20+ buf.build/gen/go/pluginrpc/pluginrpc/protocolbuffers/go v1.36.12-20241007202033-cf42259fcbfc.2 // indirect
21+ buf.build/go/app v0.2.1-0.20260824172350-b0e76892c61a // indirect
22+ buf.build/go/bufplugin v0.10.0 // indirect
23+ buf.build/go/bufprivateusage v0.1.0 // indirect
24+ buf.build/go/interrupt v1.1.0 // indirect
25+ buf.build/go/protovalidate v1.4.0 // indirect
26+ buf.build/go/protoyaml v0.7.0 // indirect
27+ buf.build/go/spdx v0.2.0 // indirect
28+ buf.build/go/standard v0.1.1-0.20260325175353-2b287e071df5 // indirect
29+ cel.dev/cel-go v0.32.0 // indirect
30+ cel.dev/expr v0.25.3 // indirect
31+ connectrpc.com/connect v1.20.0 // indirect
32+ connectrpc.com/otelconnect v0.9.0 // indirect
1533 github.com/Microsoft/go-winio v0.6.2 // indirect
34+ github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
35+ github.com/bufbuild/buf v1.73.0 // indirect
36+ github.com/bufbuild/protocompile v0.14.2-0.20260910151042-7436f7c76201 // indirect
37+ github.com/bufbuild/protoplugin v0.0.0-20260414125817-25d1d281b46b // indirect
1638 github.com/cespare/xxhash/v2 v2.3.0 // indirect
39+ github.com/cli/browser v1.3.0 // indirect
1740 github.com/containerd/errdefs/pkg v0.3.0 // indirect
41+ github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
1842 github.com/distribution/reference v0.6.0 // indirect
43+ github.com/docker/cli v29.8.0+incompatible // indirect
44+ github.com/docker/docker-credential-helpers v0.9.9 // indirect
1945 github.com/docker/go-connections v0.8.2 // indirect
2046 github.com/docker/go-units v0.5.0 // indirect
21- github.com/felixge/httpsnoop v1.0.4 // indirect
47+ github.com/felixge/httpsnoop v1.1.0 // indirect
2248 github.com/go-logr/logr v1.4.4 // indirect
2349 github.com/go-logr/stdr v1.2.2 // indirect
50+ github.com/gofrs/flock v0.13.1 // indirect
51+ github.com/google/go-containerregistry v0.22.1 // indirect
52+ github.com/google/uuid v1.6.0 // indirect
53+ github.com/inconshreveable/mousetrap v1.1.0 // indirect
54+ github.com/jdx/go-netrc v1.0.0 // indirect
55+ github.com/klauspost/compress v1.20.0 // indirect
56+ github.com/klauspost/pgzip v1.2.6 // indirect
2457 github.com/mattn/go-colorable v0.1.15 // indirect
25- github.com/mattn/go-isatty v0.0.22 // indirect
58+ github.com/mattn/go-isatty v0.0.24 // indirect
2659 github.com/moby/docker-image-spec v1.3.1 // indirect
2760 github.com/opencontainers/go-digest v1.0.0 // indirect
2861 github.com/opencontainers/image-spec v1.1.1 // indirect
62+ github.com/petermattis/goid v0.0.0-20260820044319-269ab09b5261 // indirect
63+ github.com/quic-go/qpack v0.6.0 // indirect
64+ github.com/quic-go/quic-go v0.62.0 // indirect
65+ github.com/rivo/uniseg v0.4.7 // indirect
66+ github.com/rs/cors v1.11.1 // indirect
2967 github.com/rs/zerolog v1.35.1 // indirect
68+ github.com/russross/blackfriday/v2 v2.1.0 // indirect
69+ github.com/segmentio/asm v1.2.1 // indirect
70+ github.com/segmentio/encoding v0.5.4 // indirect
71+ github.com/sirupsen/logrus v1.10.2 // indirect
72+ github.com/spf13/cobra v1.10.2 // indirect
73+ github.com/spf13/pflag v1.0.10 // indirect
74+ github.com/tetratelabs/wazero v1.12.0 // indirect
75+ github.com/tidwall/btree v1.8.1 // indirect
76+ go.lsp.dev/jsonrpc2 v0.10.0 // indirect
77+ go.lsp.dev/pkg v0.0.0-20210717090340-384b27a52fb2 // indirect
78+ go.lsp.dev/protocol v0.12.0 // indirect
79+ go.lsp.dev/uri v0.3.0 // indirect
3080 go.opentelemetry.io/auto/sdk v1.2.1 // indirect
31- go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
32- go.opentelemetry.io/otel v1.45.0 // indirect
33- go.opentelemetry.io/otel/metric v1.45.0 // indirect
34- go.opentelemetry.io/otel/trace v1.45.0 // indirect
81+ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.71.0 // indirect
82+ go.opentelemetry.io/otel v1.46.0 // indirect
83+ go.opentelemetry.io/otel/metric v1.46.0 // indirect
84+ go.opentelemetry.io/otel/trace v1.46.0 // indirect
85+ go.uber.org/multierr v1.11.0 // indirect
86+ go.uber.org/zap v1.28.0 // indirect
87+ go.yaml.in/yaml/v3 v3.0.5 // indirect
88+ golang.org/x/crypto v0.56.0 // indirect
89+ golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect
90+ golang.org/x/mod v0.40.0 // indirect
91+ golang.org/x/net v0.58.0 // indirect
92+ golang.org/x/sync v0.23.0 // indirect
93+ golang.org/x/term v0.45.0 // indirect
94+ golang.org/x/text v0.41.0 // indirect
95+ google.golang.org/genproto/googleapis/api v0.0.0-20260904194346-d0f1323225a4 // indirect
96+ google.golang.org/genproto/googleapis/rpc v0.0.0-20260904194346-d0f1323225a4 // indirect
3597 gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
98+ gopkg.in/yaml.v3 v3.0.1 // indirect
99+ mvdan.cc/xurls/v2 v2.6.0 // indirect
100+ pluginrpc.com/pluginrpc v0.5.0 // indirect
101+)
102+
103+tool (
104+ github.com/bufbuild/buf/cmd/buf
105+ google.golang.org/protobuf/cmd/protoc-gen-go
36106 )
@@ -12,25 +12,95 @@ require (
12 )12 )
13 13
14 require (14 require (
15+ buf.build/gen/go/bufbuild/bufplugin/protocolbuffers/go v1.36.12-20260722160903-4d94f3df3a7b.2 // indirect
16+ buf.build/gen/go/bufbuild/protodescriptor/protocolbuffers/go v1.36.12-20250109164928-1da0de137947.2 // indirect
17+ buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2 // indirect
18+ buf.build/gen/go/bufbuild/registry/connectrpc/go v1.20.0-20260831211851-b4432e12a6e7.1 // indirect
19+ buf.build/gen/go/bufbuild/registry/protocolbuffers/go v1.36.12-20260831211851-b4432e12a6e7.2 // indirect
20+ buf.build/gen/go/pluginrpc/pluginrpc/protocolbuffers/go v1.36.12-20241007202033-cf42259fcbfc.2 // indirect
21+ buf.build/go/app v0.2.1-0.20260824172350-b0e76892c61a // indirect
22+ buf.build/go/bufplugin v0.10.0 // indirect
23+ buf.build/go/bufprivateusage v0.1.0 // indirect
24+ buf.build/go/interrupt v1.1.0 // indirect
25+ buf.build/go/protovalidate v1.4.0 // indirect
26+ buf.build/go/protoyaml v0.7.0 // indirect
27+ buf.build/go/spdx v0.2.0 // indirect
28+ buf.build/go/standard v0.1.1-0.20260325175353-2b287e071df5 // indirect
29+ cel.dev/cel-go v0.32.0 // indirect
30+ cel.dev/expr v0.25.3 // indirect
31+ connectrpc.com/connect v1.20.0 // indirect
32+ connectrpc.com/otelconnect v0.9.0 // indirect
15 github.com/Microsoft/go-winio v0.6.2 // indirect33 github.com/Microsoft/go-winio v0.6.2 // indirect
34+ github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
35+ github.com/bufbuild/buf v1.73.0 // indirect
36+ github.com/bufbuild/protocompile v0.14.2-0.20260910151042-7436f7c76201 // indirect
37+ github.com/bufbuild/protoplugin v0.0.0-20260414125817-25d1d281b46b // indirect
16 github.com/cespare/xxhash/v2 v2.3.0 // indirect38 github.com/cespare/xxhash/v2 v2.3.0 // indirect
39+ github.com/cli/browser v1.3.0 // indirect
17 github.com/containerd/errdefs/pkg v0.3.0 // indirect40 github.com/containerd/errdefs/pkg v0.3.0 // indirect
41+ github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
18 github.com/distribution/reference v0.6.0 // indirect42 github.com/distribution/reference v0.6.0 // indirect
43+ github.com/docker/cli v29.8.0+incompatible // indirect
44+ github.com/docker/docker-credential-helpers v0.9.9 // indirect
19 github.com/docker/go-connections v0.8.2 // indirect45 github.com/docker/go-connections v0.8.2 // indirect
20 github.com/docker/go-units v0.5.0 // indirect46 github.com/docker/go-units v0.5.0 // indirect
21- github.com/felixge/httpsnoop v1.0.4 // indirect47+ github.com/felixge/httpsnoop v1.1.0 // indirect
22 github.com/go-logr/logr v1.4.4 // indirect48 github.com/go-logr/logr v1.4.4 // indirect
23 github.com/go-logr/stdr v1.2.2 // indirect49 github.com/go-logr/stdr v1.2.2 // indirect
50+ github.com/gofrs/flock v0.13.1 // indirect
51+ github.com/google/go-containerregistry v0.22.1 // indirect
52+ github.com/google/uuid v1.6.0 // indirect
53+ github.com/inconshreveable/mousetrap v1.1.0 // indirect
54+ github.com/jdx/go-netrc v1.0.0 // indirect
55+ github.com/klauspost/compress v1.20.0 // indirect
56+ github.com/klauspost/pgzip v1.2.6 // indirect
24 github.com/mattn/go-colorable v0.1.15 // indirect57 github.com/mattn/go-colorable v0.1.15 // indirect
25- github.com/mattn/go-isatty v0.0.22 // indirect58+ github.com/mattn/go-isatty v0.0.24 // indirect
26 github.com/moby/docker-image-spec v1.3.1 // indirect59 github.com/moby/docker-image-spec v1.3.1 // indirect
27 github.com/opencontainers/go-digest v1.0.0 // indirect60 github.com/opencontainers/go-digest v1.0.0 // indirect
28 github.com/opencontainers/image-spec v1.1.1 // indirect61 github.com/opencontainers/image-spec v1.1.1 // indirect
62+ github.com/petermattis/goid v0.0.0-20260820044319-269ab09b5261 // indirect
63+ github.com/quic-go/qpack v0.6.0 // indirect
64+ github.com/quic-go/quic-go v0.62.0 // indirect
65+ github.com/rivo/uniseg v0.4.7 // indirect
66+ github.com/rs/cors v1.11.1 // indirect
29 github.com/rs/zerolog v1.35.1 // indirect67 github.com/rs/zerolog v1.35.1 // indirect
68+ github.com/russross/blackfriday/v2 v2.1.0 // indirect
69+ github.com/segmentio/asm v1.2.1 // indirect
70+ github.com/segmentio/encoding v0.5.4 // indirect
71+ github.com/sirupsen/logrus v1.10.2 // indirect
72+ github.com/spf13/cobra v1.10.2 // indirect
73+ github.com/spf13/pflag v1.0.10 // indirect
74+ github.com/tetratelabs/wazero v1.12.0 // indirect
75+ github.com/tidwall/btree v1.8.1 // indirect
76+ go.lsp.dev/jsonrpc2 v0.10.0 // indirect
77+ go.lsp.dev/pkg v0.0.0-20210717090340-384b27a52fb2 // indirect
78+ go.lsp.dev/protocol v0.12.0 // indirect
79+ go.lsp.dev/uri v0.3.0 // indirect
30 go.opentelemetry.io/auto/sdk v1.2.1 // indirect80 go.opentelemetry.io/auto/sdk v1.2.1 // indirect
31- go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect81+ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.71.0 // indirect
32- go.opentelemetry.io/otel v1.45.0 // indirect82+ go.opentelemetry.io/otel v1.46.0 // indirect
33- go.opentelemetry.io/otel/metric v1.45.0 // indirect83+ go.opentelemetry.io/otel/metric v1.46.0 // indirect
34- go.opentelemetry.io/otel/trace v1.45.0 // indirect84+ go.opentelemetry.io/otel/trace v1.46.0 // indirect
85+ go.uber.org/multierr v1.11.0 // indirect
86+ go.uber.org/zap v1.28.0 // indirect
87+ go.yaml.in/yaml/v3 v3.0.5 // indirect
88+ golang.org/x/crypto v0.56.0 // indirect
89+ golang.org/x/exp v0.0.0-20260824195058-e88cd73687aa // indirect
90+ golang.org/x/mod v0.40.0 // indirect
91+ golang.org/x/net v0.58.0 // indirect
92+ golang.org/x/sync v0.23.0 // indirect
93+ golang.org/x/term v0.45.0 // indirect
94+ golang.org/x/text v0.41.0 // indirect
95+ google.golang.org/genproto/googleapis/api v0.0.0-20260904194346-d0f1323225a4 // indirect
96+ google.golang.org/genproto/googleapis/rpc v0.0.0-20260904194346-d0f1323225a4 // indirect
35 gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect97 gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
98+ gopkg.in/yaml.v3 v3.0.1 // indirect
99+ mvdan.cc/xurls/v2 v2.6.0 // indirect
100+ pluginrpc.com/pluginrpc v0.5.0 // indirect
101+)
102+
103+tool (
104+ github.com/bufbuild/buf/cmd/buf
105+ google.golang.org/protobuf/cmd/protoc-gen-go
36 )106 )
M go.sum +178 -20 generated file: collapsed
load diff
M internal/execstream/exec.proto +8 -2
@@ -54,5 +54,11 @@ message Exit {
5454 string error = 2; // set when the executor failed to run the script at all
5555 }
5656
57-// Cancel asks the executor to stop the running script: SIGINT now.
58-message Cancel {}
57+// Cancel asks the executor to stop the running script. The executor escalates
58+// like the official runner: SIGINT to the script's process group at once,
59+// SIGTERM 7.5 s later, SIGKILL 2.5 s after that.
60+message Cancel {
61+ // reason is for the job log and diagnostics ("cancelled by the user",
62+ // "job timeout", ...); it never changes the behaviour.
63+ string reason = 1;
64+}
@@ -54,5 +54,11 @@ message Exit {
54 string error = 2; // set when the executor failed to run the script at all54 string error = 2; // set when the executor failed to run the script at all
55 }55 }
56 56
57-// Cancel asks the executor to stop the running script: SIGINT now.57+// Cancel asks the executor to stop the running script. The executor escalates
58-message Cancel {}58+// like the official runner: SIGINT to the script's process group at once,
59+// SIGTERM 7.5 s later, SIGKILL 2.5 s after that.
60+message Cancel {
61+ // reason is for the job log and diagnostics ("cancelled by the user",
62+ // "job timeout", ...); it never changes the behaviour.
63+ string reason = 1;
64+}
M internal/execstream/execpb/exec.pb.go +19 -6
@@ -9,8 +9,8 @@
99
1010 // Code generated by protoc-gen-go. DO NOT EDIT.
1111 // versions:
12-// protoc-gen-go v1.36.11
13-// protoc v7.36.1
12+// protoc-gen-go v1.36.12
13+// protoc (unknown)
1414 // source: exec.proto
1515
1616 package execpb
@@ -436,9 +436,14 @@ func (x *Exit) GetError() string {
436436 return ""
437437 }
438438
439-// Cancel asks the executor to stop the running script: SIGINT now.
439+// Cancel asks the executor to stop the running script. The executor escalates
440+// like the official runner: SIGINT to the script's process group at once,
441+// SIGTERM 7.5 s later, SIGKILL 2.5 s after that.
440442 type Cancel struct {
441- state protoimpl.MessageState `protogen:"open.v1"`
443+ state protoimpl.MessageState `protogen:"open.v1"`
444+ // reason is for the job log and diagnostics ("cancelled by the user",
445+ // "job timeout", ...); it never changes the behaviour.
446+ Reason string `protobuf:"bytes,1,opt,name=reason,proto3" json:"reason,omitempty"`
442447 unknownFields protoimpl.UnknownFields
443448 sizeCache protoimpl.SizeCache
444449 }
@@ -473,6 +478,13 @@ func (*Cancel) Descriptor() ([]byte, []int) {
473478 return file_exec_proto_rawDescGZIP(), []int{5}
474479 }
475480
481+func (x *Cancel) GetReason() string {
482+ if x != nil {
483+ return x.Reason
484+ }
485+ return ""
486+}
487+
476488 var File_exec_proto protoreflect.FileDescriptor
477489
478490 const file_exec_proto_rawDesc = "" +
@@ -502,8 +514,9 @@ const file_exec_proto_rawDesc = "" +
502514 "\tunix_nano\x18\x03 \x01(\x03R\bunixNano\"0\n" +
503515 "\x04Exit\x12\x12\n" +
504516 "\x04code\x18\x01 \x01(\x05R\x04code\x12\x14\n" +
505- "\x05error\x18\x02 \x01(\tR\x05error\"\b\n" +
506- "\x06Cancel*Y\n" +
517+ "\x05error\x18\x02 \x01(\tR\x05error\" \n" +
518+ "\x06Cancel\x12\x16\n" +
519+ "\x06reason\x18\x01 \x01(\tR\x06reason*Y\n" +
507520 "\n" +
508521 "StreamKind\x12\x1b\n" +
509522 "\x17STREAM_KIND_UNSPECIFIED\x10\x00\x12\x16\n" +
@@ -9,8 +9,8 @@
9 9
10 // Code generated by protoc-gen-go. DO NOT EDIT.10 // Code generated by protoc-gen-go. DO NOT EDIT.
11 // versions:11 // versions:
12-// protoc-gen-go v1.36.1112+// protoc-gen-go v1.36.12
13-// protoc v7.36.113+// protoc (unknown)
14 // source: exec.proto14 // source: exec.proto
15 15
16 package execpb16 package execpb
@@ -436,9 +436,14 @@ func (x *Exit) GetError() string {
436 return ""436 return ""
437 }437 }
438 438
439-// Cancel asks the executor to stop the running script: SIGINT now.439+// Cancel asks the executor to stop the running script. The executor escalates
440+// like the official runner: SIGINT to the script's process group at once,
441+// SIGTERM 7.5 s later, SIGKILL 2.5 s after that.
440 type Cancel struct {442 type Cancel struct {
441- state protoimpl.MessageState `protogen:"open.v1"`443+ state protoimpl.MessageState `protogen:"open.v1"`
444+ // reason is for the job log and diagnostics ("cancelled by the user",
445+ // "job timeout", ...); it never changes the behaviour.
446+ Reason string `protobuf:"bytes,1,opt,name=reason,proto3" json:"reason,omitempty"`
442 unknownFields protoimpl.UnknownFields447 unknownFields protoimpl.UnknownFields
443 sizeCache protoimpl.SizeCache448 sizeCache protoimpl.SizeCache
444 }449 }
@@ -473,6 +478,13 @@ func (*Cancel) Descriptor() ([]byte, []int) {
473 return file_exec_proto_rawDescGZIP(), []int{5}478 return file_exec_proto_rawDescGZIP(), []int{5}
474 }479 }
475 480
481+func (x *Cancel) GetReason() string {
482+ if x != nil {
483+ return x.Reason
484+ }
485+ return ""
486+}
487+
476 var File_exec_proto protoreflect.FileDescriptor488 var File_exec_proto protoreflect.FileDescriptor
477 489
478 const file_exec_proto_rawDesc = "" +490 const file_exec_proto_rawDesc = "" +
@@ -502,8 +514,9 @@ const file_exec_proto_rawDesc = "" +
502 "\tunix_nano\x18\x03 \x01(\x03R\bunixNano\"0\n" +514 "\tunix_nano\x18\x03 \x01(\x03R\bunixNano\"0\n" +
503 "\x04Exit\x12\x12\n" +515 "\x04Exit\x12\x12\n" +
504 "\x04code\x18\x01 \x01(\x05R\x04code\x12\x14\n" +516 "\x04code\x18\x01 \x01(\x05R\x04code\x12\x14\n" +
505- "\x05error\x18\x02 \x01(\tR\x05error\"\b\n" +517+ "\x05error\x18\x02 \x01(\tR\x05error\" \n" +
506- "\x06Cancel*Y\n" +518+ "\x06Cancel\x12\x16\n" +
519+ "\x06reason\x18\x01 \x01(\tR\x06reason*Y\n" +
507 "\n" +520 "\n" +
508 "StreamKind\x12\x1b\n" +521 "StreamKind\x12\x1b\n" +
509 "\x17STREAM_KIND_UNSPECIFIED\x10\x00\x12\x16\n" +522 "\x17STREAM_KIND_UNSPECIFIED\x10\x00\x12\x16\n" +
M internal/execstream/frame.go +24 -13
@@ -9,11 +9,10 @@
99 // transport-agnostic: vsock in a micro-VM, a Unix socket in a container.
1010 package execstream
1111
12-//go:generate protoc --proto_path=. --go_out=execpb --go_opt=paths=source_relative exec.proto
12+//go:generate go tool buf generate ../.. --template ../../buf.gen.yaml
1313
1414 import (
1515 "bufio"
16- "context"
1716 "encoding/binary"
1817 "errors"
1918 "fmt"
@@ -92,10 +91,15 @@ func (c *Conn) Recv() (*execpb.Frame, error) {
9291
9392 // Drive is the host side of a session: it waits for the executor's Hello, sends
9493 // run, passes every Output frame to onOutput, and returns the Exit frame.
95-// When ctx ends first, a Cancel frame is sent once and Drive keeps waiting for
96-// the executor's Exit; the caller bounds that wait by stopping the sandbox,
97-// which closes the stream and makes Drive return an error.
98-func Drive(ctx context.Context, rw io.ReadWriter, run *execpb.Run, onOutput func(*execpb.Output)) (*execpb.Hello, *execpb.Exit, error) {
94+//
95+// cancel, when non-nil, asks the executor to stop the script: the first value
96+// received is sent as one Cancel frame (the executor escalates SIGINT, SIGTERM,
97+// SIGKILL by itself); closing the channel without a value sends nothing. It is
98+// deliberately not a context: cancelling the session's context would also abort
99+// the reads that deliver the executor's Exit. Drive keeps waiting for Exit after
100+// a Cancel; the caller bounds that wait by stopping the sandbox, which closes
101+// the stream and makes Drive return an error.
102+func Drive(rw io.ReadWriter, run *execpb.Run, cancel <-chan *execpb.Cancel, onOutput func(*execpb.Output)) (*execpb.Hello, *execpb.Exit, error) {
99103 c := NewConn(rw)
100104 first, err := c.Recv()
101105 if err != nil {
@@ -111,13 +115,20 @@ func Drive(ctx context.Context, rw io.ReadWriter, run *execpb.Run, onOutput func
111115
112116 done := make(chan struct{})
113117 defer close(done)
114- go func() {
115- select {
116- case <-ctx.Done():
117- _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Cancel{Cancel: &execpb.Cancel{}}})
118- case <-done:
119- }
120- }()
118+ if cancel != nil {
119+ go func() {
120+ select {
121+ case cn, ok := <-cancel:
122+ if ok {
123+ if cn == nil {
124+ cn = &execpb.Cancel{}
125+ }
126+ _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Cancel{Cancel: cn}})
127+ }
128+ case <-done:
129+ }
130+ }()
131+ }
121132
122133 for {
123134 f, err := c.Recv()
@@ -9,11 +9,10 @@
9 // transport-agnostic: vsock in a micro-VM, a Unix socket in a container.9 // transport-agnostic: vsock in a micro-VM, a Unix socket in a container.
10 package execstream10 package execstream
11 11
12-//go:generate protoc --proto_path=. --go_out=execpb --go_opt=paths=source_relative exec.proto12+//go:generate go tool buf generate ../.. --template ../../buf.gen.yaml
13 13
14 import (14 import (
15 "bufio"15 "bufio"
16- "context"
17 "encoding/binary"16 "encoding/binary"
18 "errors"17 "errors"
19 "fmt"18 "fmt"
@@ -92,10 +91,15 @@ func (c *Conn) Recv() (*execpb.Frame, error) {
92 91
93 // Drive is the host side of a session: it waits for the executor's Hello, sends92 // Drive is the host side of a session: it waits for the executor's Hello, sends
94 // run, passes every Output frame to onOutput, and returns the Exit frame.93 // run, passes every Output frame to onOutput, and returns the Exit frame.
95-// When ctx ends first, a Cancel frame is sent once and Drive keeps waiting for94+//
96-// the executor's Exit; the caller bounds that wait by stopping the sandbox,95+// cancel, when non-nil, asks the executor to stop the script: the first value
97-// which closes the stream and makes Drive return an error.96+// received is sent as one Cancel frame (the executor escalates SIGINT, SIGTERM,
98-func Drive(ctx context.Context, rw io.ReadWriter, run *execpb.Run, onOutput func(*execpb.Output)) (*execpb.Hello, *execpb.Exit, error) {97+// SIGKILL by itself); closing the channel without a value sends nothing. It is
98+// deliberately not a context: cancelling the session's context would also abort
99+// the reads that deliver the executor's Exit. Drive keeps waiting for Exit after
100+// a Cancel; the caller bounds that wait by stopping the sandbox, which closes
101+// the stream and makes Drive return an error.
102+func Drive(rw io.ReadWriter, run *execpb.Run, cancel <-chan *execpb.Cancel, onOutput func(*execpb.Output)) (*execpb.Hello, *execpb.Exit, error) {
99 c := NewConn(rw)103 c := NewConn(rw)
100 first, err := c.Recv()104 first, err := c.Recv()
101 if err != nil {105 if err != nil {
@@ -111,13 +115,20 @@ func Drive(ctx context.Context, rw io.ReadWriter, run *execpb.Run, onOutput func
111 115
112 done := make(chan struct{})116 done := make(chan struct{})
113 defer close(done)117 defer close(done)
114- go func() {118+ if cancel != nil {
115- select {119+ go func() {
116- case <-ctx.Done():120+ select {
117- _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Cancel{Cancel: &execpb.Cancel{}}})121+ case cn, ok := <-cancel:
118- case <-done:122+ if ok {
119- }123+ if cn == nil {
120- }()124+ cn = &execpb.Cancel{}
125+ }
126+ _ = c.Send(&execpb.Frame{Body: &execpb.Frame_Cancel{Cancel: cn}})
127+ }
128+ case <-done:
129+ }
130+ }()
131+ }
121 132
122 for {133 for {
123 f, err := c.Recv()134 f, err := c.Recv()
M internal/execstream/frame_test.go +6 -7
@@ -4,7 +4,6 @@ package execstream
44
55 import (
66 "bytes"
7- "context"
87 "encoding/binary"
98 "errors"
109 "io"
@@ -93,7 +92,7 @@ func TestDrive(t *testing.T) {
9392 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})
9493 }()
9594 var got []byte
96- hello, exit, err := Drive(context.Background(), host, &execpb.Run{Script: "echo hi"}, func(o *execpb.Output) { got = append(got, o.Data...) })
95+ hello, exit, err := Drive(host, &execpb.Run{Script: "echo hi"}, nil, func(o *execpb.Output) { got = append(got, o.Data...) })
9796 if err != nil {
9897 t.Fatal(err)
9998 }
@@ -112,12 +111,12 @@ func TestDriveCancelSendsCancel(t *testing.T) {
112111 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})
113112 _, _ = g.Recv() // run
114113 f, err := g.Recv()
115- gotCancel <- err == nil && f.GetCancel() != nil
114+ gotCancel <- err == nil && f.GetCancel() != nil && f.GetCancel().Reason == "cancelled by the user"
116115 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 130}}})
117116 }()
118- ctx, cancel := context.WithCancel(context.Background())
119- time.AfterFunc(50*time.Millisecond, cancel)
120- _, exit, err := Drive(ctx, host, &execpb.Run{}, nil)
117+ cancel := make(chan *execpb.Cancel, 1)
118+ time.AfterFunc(50*time.Millisecond, func() { cancel <- &execpb.Cancel{Reason: "cancelled by the user"} })
119+ _, exit, err := Drive(host, &execpb.Run{}, cancel, nil)
121120 if err != nil || exit.Code != 130 {
122121 t.Fatalf("exit=%v err=%v", exit, err)
123122 }
@@ -133,7 +132,7 @@ func TestDriveHelloMissing(t *testing.T) {
133132 defer func() { _ = guest.Close() }()
134133 _ = NewConn(guest).Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
135134 }()
136- if _, _, err := Drive(context.Background(), host, &execpb.Run{}, nil); err == nil {
135+ if _, _, err := Drive(host, &execpb.Run{}, nil, nil); err == nil {
137136 t.Fatal("want error when the first frame is not hello")
138137 }
139138 }
@@ -4,7 +4,6 @@ package execstream
4 4
5 import (5 import (
6 "bytes"6 "bytes"
7- "context"
8 "encoding/binary"7 "encoding/binary"
9 "errors"8 "errors"
10 "io"9 "io"
@@ -93,7 +92,7 @@ func TestDrive(t *testing.T) {
93 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})92 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})
94 }()93 }()
95 var got []byte94 var got []byte
96- hello, exit, err := Drive(context.Background(), host, &execpb.Run{Script: "echo hi"}, func(o *execpb.Output) { got = append(got, o.Data...) })95+ hello, exit, err := Drive(host, &execpb.Run{Script: "echo hi"}, nil, func(o *execpb.Output) { got = append(got, o.Data...) })
97 if err != nil {96 if err != nil {
98 t.Fatal(err)97 t.Fatal(err)
99 }98 }
@@ -112,12 +111,12 @@ func TestDriveCancelSendsCancel(t *testing.T) {
112 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})111 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{}}})
113 _, _ = g.Recv() // run112 _, _ = g.Recv() // run
114 f, err := g.Recv()113 f, err := g.Recv()
115- gotCancel <- err == nil && f.GetCancel() != nil114+ gotCancel <- err == nil && f.GetCancel() != nil && f.GetCancel().Reason == "cancelled by the user"
116 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 130}}})115 _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 130}}})
117 }()116 }()
118- ctx, cancel := context.WithCancel(context.Background())117+ cancel := make(chan *execpb.Cancel, 1)
119- time.AfterFunc(50*time.Millisecond, cancel)118+ time.AfterFunc(50*time.Millisecond, func() { cancel <- &execpb.Cancel{Reason: "cancelled by the user"} })
120- _, exit, err := Drive(ctx, host, &execpb.Run{}, nil)119+ _, exit, err := Drive(host, &execpb.Run{}, cancel, nil)
121 if err != nil || exit.Code != 130 {120 if err != nil || exit.Code != 130 {
122 t.Fatalf("exit=%v err=%v", exit, err)121 t.Fatalf("exit=%v err=%v", exit, err)
123 }122 }
@@ -133,7 +132,7 @@ func TestDriveHelloMissing(t *testing.T) {
133 defer func() { _ = guest.Close() }()132 defer func() { _ = guest.Close() }()
134 _ = NewConn(guest).Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})133 _ = NewConn(guest).Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}})
135 }()134 }()
136- if _, _, err := Drive(context.Background(), host, &execpb.Run{}, nil); err == nil {135+ if _, _, err := Drive(host, &execpb.Run{}, nil, nil); err == nil {
137 t.Fatal("want error when the first frame is not hello")136 t.Fatal("want error when the first frame is not hello")
138 }137 }
139 }138 }
M internal/guest/exec.go +144 -43
@@ -20,6 +20,17 @@ import (
2020 // DefaultScript is what the stub executor runs when the host sends no script.
2121 const DefaultScript = "uname -a; id; cat /etc/os-release; docker info"
2222
23+// Cancellation timings, like the official runner's ProcessInvoker: SIGINT, then
24+// SIGTERM after 7.5 s, then SIGKILL 2.5 s later.
25+const (
26+ DefaultTermAfter = 7500 * time.Millisecond
27+ DefaultKillAfter = 2500 * time.Millisecond
28+ // DefaultWaitDelay bounds how long the executor waits for the output pipes
29+ // to close after the script's shell exited (background children may hold
30+ // them open).
31+ DefaultWaitDelay = 500 * time.Millisecond
32+)
33+
2334 // ExecConfig configures the executor.
2435 type ExecConfig struct {
2536 Target Target
@@ -31,22 +42,48 @@ type ExecConfig struct {
3142 ConnectTimeout time.Duration
3243 // RunTimeout bounds the wait for the host's Run frame (default 60 s).
3344 RunTimeout time.Duration
45+ // TermAfter and KillAfter are the escalation delays after a Cancel (or after
46+ // the executor's own context ends): SIGINT at once, SIGTERM after TermAfter,
47+ // SIGKILL KillAfter later. WaitDelay is the pipe-drain grace after the shell
48+ // exits. Zero means the defaults above.
49+ TermAfter, KillAfter, WaitDelay time.Duration
50+
51+ // signal sends a signal to a process group; tests replace it to observe the
52+ // escalation. Default: syscall.Kill(-pgid, sig).
53+ signal func(pgid int, sig syscall.Signal)
54+}
55+
56+func (c *ExecConfig) defaults() {
57+ if c.Shell == "" {
58+ c.Shell = "/bin/sh"
59+ }
60+ if c.ConnectTimeout == 0 {
61+ c.ConnectTimeout = 30 * time.Second
62+ }
63+ if c.RunTimeout == 0 {
64+ c.RunTimeout = 60 * time.Second
65+ }
66+ if c.TermAfter == 0 {
67+ c.TermAfter = DefaultTermAfter
68+ }
69+ if c.KillAfter == 0 {
70+ c.KillAfter = DefaultKillAfter
71+ }
72+ if c.WaitDelay == 0 {
73+ c.WaitDelay = DefaultWaitDelay
74+ }
75+ if c.signal == nil {
76+ c.signal = func(pgid int, sig syscall.Signal) { _ = syscall.Kill(-pgid, sig) }
77+ }
3478 }
3579
3680 // Exec is `ermite-runner exec`, the CI-1 stub executor: it connects back to the
3781 // host, receives one Run, runs the script with the shell, streams stdout and
3882 // stderr as Output frames, and ends with an Exit frame. It returns the exit code
39-// it reported. A Cancel frame sends SIGINT to the script's process group.
83+// it reported. A Cancel frame, or the end of ctx, stops the script's process
84+// group with escalating signals (see ExecConfig).
4085 func Exec(ctx context.Context, cfg ExecConfig) (int, error) {
41- if cfg.Shell == "" {
42- cfg.Shell = "/bin/sh"
43- }
44- if cfg.ConnectTimeout == 0 {
45- cfg.ConnectTimeout = 30 * time.Second
46- }
47- if cfg.RunTimeout == 0 {
48- cfg.RunTimeout = 60 * time.Second
49- }
86+ cfg.defaults()
5087 raw, err := cfg.Target.Dial(ctx, cfg.ConnectTimeout)
5188 if err != nil {
5289 return 1, err
@@ -55,6 +92,85 @@ func Exec(ctx context.Context, cfg ExecConfig) (int, error) {
5592 return serve(ctx, raw, cfg)
5693 }
5794
95+// frameWriter turns writes into Output frames of at most ChunkSize bytes. It is
96+// the command's Stdout/Stderr, so no goroutine of ours sits between the pipe
97+// and the stream.
98+type frameWriter struct {
99+ conn *execstream.Conn
100+ kind execpb.StreamKind
101+}
102+
103+func (w frameWriter) Write(p []byte) (int, error) {
104+ n := 0
105+ for len(p) > 0 {
106+ chunk := p
107+ if len(chunk) > execstream.ChunkSize {
108+ chunk = chunk[:execstream.ChunkSize]
109+ }
110+ err := w.conn.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{
111+ Stream: w.kind, Data: append([]byte(nil), chunk...), UnixNano: time.Now().UnixNano(),
112+ }}})
113+ if err != nil {
114+ return n, err
115+ }
116+ n += len(chunk)
117+ p = p[len(chunk):]
118+ }
119+ return n, nil
120+}
121+
122+// escalator signals a process group and escalates after a cancel. Once finish
123+// has been called (the shell was reaped) it never signals again: the group id
124+// could have been reused by then.
125+type escalator struct {
126+ cfg ExecConfig
127+ pgid int
128+
129+ mu sync.Mutex
130+ finished bool
131+ started bool
132+ timers []*time.Timer
133+}
134+
135+func (e *escalator) send(sig syscall.Signal) {
136+ e.mu.Lock()
137+ defer e.mu.Unlock()
138+ if !e.finished {
139+ e.cfg.signal(e.pgid, sig)
140+ }
141+}
142+
143+// cancel starts the escalation; only the first call does anything.
144+func (e *escalator) cancel() {
145+ e.mu.Lock()
146+ if e.finished || e.started {
147+ e.mu.Unlock()
148+ return
149+ }
150+ e.started = true
151+ e.cfg.signal(e.pgid, syscall.SIGINT)
152+ e.timers = append(e.timers,
153+ time.AfterFunc(e.cfg.TermAfter, func() { e.send(syscall.SIGTERM) }),
154+ time.AfterFunc(e.cfg.TermAfter+e.cfg.KillAfter, func() { e.send(syscall.SIGKILL) }),
155+ )
156+ e.mu.Unlock()
157+}
158+
159+// finish is called right after the shell was reaped: it kills what is left of
160+// the group (background children), stops the timers and forbids further signals.
161+func (e *escalator) finish() {
162+ e.mu.Lock()
163+ defer e.mu.Unlock()
164+ if e.finished {
165+ return
166+ }
167+ e.cfg.signal(e.pgid, syscall.SIGKILL)
168+ e.finished = true
169+ for _, t := range e.timers {
170+ t.Stop()
171+ }
172+}
173+
58174 func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, error) {
59175 conn := execstream.NewConn(rw)
60176 host, _ := os.Hostname()
@@ -96,58 +212,43 @@ func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, err
96212 cmd.Env = append(cmd.Env, k+"="+v)
97213 }
98214 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
99- stdout, err := cmd.StdoutPipe()
100- if err != nil {
101- return sendExit(conn, 1, err)
102- }
103- stderr, err := cmd.StderrPipe()
104- if err != nil {
105- return sendExit(conn, 1, err)
106- }
215+ // Frame writers as stdout/stderr: exec copies the pipes for us, and
216+ // WaitDelay bounds the wait when a background child keeps a pipe open.
217+ cmd.Stdout = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDOUT}
218+ cmd.Stderr = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDERR}
219+ cmd.WaitDelay = cfg.WaitDelay
107220 if err := cmd.Start(); err != nil {
108221 return sendExit(conn, 127, fmt.Errorf("start %s: %w", cfg.Shell, err))
109222 }
223+ esc := &escalator{cfg: cfg, pgid: cmd.Process.Pid}
110224
111- // Cancel frames arrive while the script runs.
225+ // Frames from the host while the script runs: a Cancel starts the escalation;
226+ // a lost host kills the script rather than let it run unobserved.
112227 go func() {
113228 for {
114229 f, err := conn.Recv()
115230 if err != nil {
116- // Host went away: stop the script rather than run unobserved.
117- _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
231+ esc.send(syscall.SIGKILL)
118232 return
119233 }
120234 if f.GetCancel() != nil {
121- _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGINT)
235+ esc.cancel()
122236 }
123237 }
124238 }()
239+ // The executor's own context (SIGTERM forwarded by init) cancels the same way.
240+ stopCtx := context.AfterFunc(ctx, esc.cancel)
241+ defer stopCtx()
125242
126- var wg sync.WaitGroup
127- pump := func(r io.Reader, kind execpb.StreamKind) {
128- defer wg.Done()
129- buf := make([]byte, execstream.ChunkSize)
130- for {
131- n, err := r.Read(buf)
132- if n > 0 {
133- _ = conn.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{
134- Stream: kind, Data: append([]byte(nil), buf[:n]...), UnixNano: time.Now().UnixNano(),
135- }}})
136- }
137- if err != nil {
138- return
139- }
140- }
141- }
142- wg.Add(2)
143- go pump(stdout, execpb.StreamKind_STREAM_KIND_STDOUT)
144- go pump(stderr, execpb.StreamKind_STREAM_KIND_STDERR)
145- wg.Wait() // pipes must be drained before Wait closes them
146243 waitErr := cmd.Wait()
244+ esc.finish()
245+
147246 code := 0
148247 var ee *exec.ExitError
149248 switch {
150- case waitErr == nil:
249+ case waitErr == nil, errors.Is(waitErr, exec.ErrWaitDelay):
250+ // ErrWaitDelay: the shell exited cleanly but a background child kept the
251+ // pipes open; the group was killed above.
151252 case errors.As(waitErr, &ee):
152253 code = ee.ExitCode()
153254 if code < 0 { // killed by a signal: 128+signal like a shell
@@ -20,6 +20,17 @@ import (
20 // DefaultScript is what the stub executor runs when the host sends no script.20 // DefaultScript is what the stub executor runs when the host sends no script.
21 const DefaultScript = "uname -a; id; cat /etc/os-release; docker info"21 const DefaultScript = "uname -a; id; cat /etc/os-release; docker info"
22 22
23+// Cancellation timings, like the official runner's ProcessInvoker: SIGINT, then
24+// SIGTERM after 7.5 s, then SIGKILL 2.5 s later.
25+const (
26+ DefaultTermAfter = 7500 * time.Millisecond
27+ DefaultKillAfter = 2500 * time.Millisecond
28+ // DefaultWaitDelay bounds how long the executor waits for the output pipes
29+ // to close after the script's shell exited (background children may hold
30+ // them open).
31+ DefaultWaitDelay = 500 * time.Millisecond
32+)
33+
23 // ExecConfig configures the executor.34 // ExecConfig configures the executor.
24 type ExecConfig struct {35 type ExecConfig struct {
25 Target Target36 Target Target
@@ -31,22 +42,48 @@ type ExecConfig struct {
31 ConnectTimeout time.Duration42 ConnectTimeout time.Duration
32 // RunTimeout bounds the wait for the host's Run frame (default 60 s).43 // RunTimeout bounds the wait for the host's Run frame (default 60 s).
33 RunTimeout time.Duration44 RunTimeout time.Duration
45+ // TermAfter and KillAfter are the escalation delays after a Cancel (or after
46+ // the executor's own context ends): SIGINT at once, SIGTERM after TermAfter,
47+ // SIGKILL KillAfter later. WaitDelay is the pipe-drain grace after the shell
48+ // exits. Zero means the defaults above.
49+ TermAfter, KillAfter, WaitDelay time.Duration
50+
51+ // signal sends a signal to a process group; tests replace it to observe the
52+ // escalation. Default: syscall.Kill(-pgid, sig).
53+ signal func(pgid int, sig syscall.Signal)
54+}
55+
56+func (c *ExecConfig) defaults() {
57+ if c.Shell == "" {
58+ c.Shell = "/bin/sh"
59+ }
60+ if c.ConnectTimeout == 0 {
61+ c.ConnectTimeout = 30 * time.Second
62+ }
63+ if c.RunTimeout == 0 {
64+ c.RunTimeout = 60 * time.Second
65+ }
66+ if c.TermAfter == 0 {
67+ c.TermAfter = DefaultTermAfter
68+ }
69+ if c.KillAfter == 0 {
70+ c.KillAfter = DefaultKillAfter
71+ }
72+ if c.WaitDelay == 0 {
73+ c.WaitDelay = DefaultWaitDelay
74+ }
75+ if c.signal == nil {
76+ c.signal = func(pgid int, sig syscall.Signal) { _ = syscall.Kill(-pgid, sig) }
77+ }
34 }78 }
35 79
36 // Exec is `ermite-runner exec`, the CI-1 stub executor: it connects back to the80 // Exec is `ermite-runner exec`, the CI-1 stub executor: it connects back to the
37 // host, receives one Run, runs the script with the shell, streams stdout and81 // host, receives one Run, runs the script with the shell, streams stdout and
38 // stderr as Output frames, and ends with an Exit frame. It returns the exit code82 // stderr as Output frames, and ends with an Exit frame. It returns the exit code
39-// it reported. A Cancel frame sends SIGINT to the script's process group.83+// it reported. A Cancel frame, or the end of ctx, stops the script's process
84+// group with escalating signals (see ExecConfig).
40 func Exec(ctx context.Context, cfg ExecConfig) (int, error) {85 func Exec(ctx context.Context, cfg ExecConfig) (int, error) {
41- if cfg.Shell == "" {86+ cfg.defaults()
42- cfg.Shell = "/bin/sh"
43- }
44- if cfg.ConnectTimeout == 0 {
45- cfg.ConnectTimeout = 30 * time.Second
46- }
47- if cfg.RunTimeout == 0 {
48- cfg.RunTimeout = 60 * time.Second
49- }
50 raw, err := cfg.Target.Dial(ctx, cfg.ConnectTimeout)87 raw, err := cfg.Target.Dial(ctx, cfg.ConnectTimeout)
51 if err != nil {88 if err != nil {
52 return 1, err89 return 1, err
@@ -55,6 +92,85 @@ func Exec(ctx context.Context, cfg ExecConfig) (int, error) {
55 return serve(ctx, raw, cfg)92 return serve(ctx, raw, cfg)
56 }93 }
57 94
95+// frameWriter turns writes into Output frames of at most ChunkSize bytes. It is
96+// the command's Stdout/Stderr, so no goroutine of ours sits between the pipe
97+// and the stream.
98+type frameWriter struct {
99+ conn *execstream.Conn
100+ kind execpb.StreamKind
101+}
102+
103+func (w frameWriter) Write(p []byte) (int, error) {
104+ n := 0
105+ for len(p) > 0 {
106+ chunk := p
107+ if len(chunk) > execstream.ChunkSize {
108+ chunk = chunk[:execstream.ChunkSize]
109+ }
110+ err := w.conn.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{
111+ Stream: w.kind, Data: append([]byte(nil), chunk...), UnixNano: time.Now().UnixNano(),
112+ }}})
113+ if err != nil {
114+ return n, err
115+ }
116+ n += len(chunk)
117+ p = p[len(chunk):]
118+ }
119+ return n, nil
120+}
121+
122+// escalator signals a process group and escalates after a cancel. Once finish
123+// has been called (the shell was reaped) it never signals again: the group id
124+// could have been reused by then.
125+type escalator struct {
126+ cfg ExecConfig
127+ pgid int
128+
129+ mu sync.Mutex
130+ finished bool
131+ started bool
132+ timers []*time.Timer
133+}
134+
135+func (e *escalator) send(sig syscall.Signal) {
136+ e.mu.Lock()
137+ defer e.mu.Unlock()
138+ if !e.finished {
139+ e.cfg.signal(e.pgid, sig)
140+ }
141+}
142+
143+// cancel starts the escalation; only the first call does anything.
144+func (e *escalator) cancel() {
145+ e.mu.Lock()
146+ if e.finished || e.started {
147+ e.mu.Unlock()
148+ return
149+ }
150+ e.started = true
151+ e.cfg.signal(e.pgid, syscall.SIGINT)
152+ e.timers = append(e.timers,
153+ time.AfterFunc(e.cfg.TermAfter, func() { e.send(syscall.SIGTERM) }),
154+ time.AfterFunc(e.cfg.TermAfter+e.cfg.KillAfter, func() { e.send(syscall.SIGKILL) }),
155+ )
156+ e.mu.Unlock()
157+}
158+
159+// finish is called right after the shell was reaped: it kills what is left of
160+// the group (background children), stops the timers and forbids further signals.
161+func (e *escalator) finish() {
162+ e.mu.Lock()
163+ defer e.mu.Unlock()
164+ if e.finished {
165+ return
166+ }
167+ e.cfg.signal(e.pgid, syscall.SIGKILL)
168+ e.finished = true
169+ for _, t := range e.timers {
170+ t.Stop()
171+ }
172+}
173+
58 func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, error) {174 func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, error) {
59 conn := execstream.NewConn(rw)175 conn := execstream.NewConn(rw)
60 host, _ := os.Hostname()176 host, _ := os.Hostname()
@@ -96,58 +212,43 @@ func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, err
96 cmd.Env = append(cmd.Env, k+"="+v)212 cmd.Env = append(cmd.Env, k+"="+v)
97 }213 }
98 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}214 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
99- stdout, err := cmd.StdoutPipe()215+ // Frame writers as stdout/stderr: exec copies the pipes for us, and
100- if err != nil {216+ // WaitDelay bounds the wait when a background child keeps a pipe open.
101- return sendExit(conn, 1, err)217+ cmd.Stdout = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDOUT}
102- }218+ cmd.Stderr = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDERR}
103- stderr, err := cmd.StderrPipe()219+ cmd.WaitDelay = cfg.WaitDelay
104- if err != nil {
105- return sendExit(conn, 1, err)
106- }
107 if err := cmd.Start(); err != nil {220 if err := cmd.Start(); err != nil {
108 return sendExit(conn, 127, fmt.Errorf("start %s: %w", cfg.Shell, err))221 return sendExit(conn, 127, fmt.Errorf("start %s: %w", cfg.Shell, err))
109 }222 }
223+ esc := &escalator{cfg: cfg, pgid: cmd.Process.Pid}
110 224
111- // Cancel frames arrive while the script runs.225+ // Frames from the host while the script runs: a Cancel starts the escalation;
226+ // a lost host kills the script rather than let it run unobserved.
112 go func() {227 go func() {
113 for {228 for {
114 f, err := conn.Recv()229 f, err := conn.Recv()
115 if err != nil {230 if err != nil {
116- // Host went away: stop the script rather than run unobserved.231+ esc.send(syscall.SIGKILL)
117- _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
118 return232 return
119 }233 }
120 if f.GetCancel() != nil {234 if f.GetCancel() != nil {
121- _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGINT)235+ esc.cancel()
122 }236 }
123 }237 }
124 }()238 }()
239+ // The executor's own context (SIGTERM forwarded by init) cancels the same way.
240+ stopCtx := context.AfterFunc(ctx, esc.cancel)
241+ defer stopCtx()
125 242
126- var wg sync.WaitGroup
127- pump := func(r io.Reader, kind execpb.StreamKind) {
128- defer wg.Done()
129- buf := make([]byte, execstream.ChunkSize)
130- for {
131- n, err := r.Read(buf)
132- if n > 0 {
133- _ = conn.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{
134- Stream: kind, Data: append([]byte(nil), buf[:n]...), UnixNano: time.Now().UnixNano(),
135- }}})
136- }
137- if err != nil {
138- return
139- }
140- }
141- }
142- wg.Add(2)
143- go pump(stdout, execpb.StreamKind_STREAM_KIND_STDOUT)
144- go pump(stderr, execpb.StreamKind_STREAM_KIND_STDERR)
145- wg.Wait() // pipes must be drained before Wait closes them
146 waitErr := cmd.Wait()243 waitErr := cmd.Wait()
244+ esc.finish()
245+
147 code := 0246 code := 0
148 var ee *exec.ExitError247 var ee *exec.ExitError
149 switch {248 switch {
150- case waitErr == nil:249+ case waitErr == nil, errors.Is(waitErr, exec.ErrWaitDelay):
250+ // ErrWaitDelay: the shell exited cleanly but a background child kept the
251+ // pipes open; the group was killed above.
151 case errors.As(waitErr, &ee):252 case errors.As(waitErr, &ee):
152 code = ee.ExitCode()253 code = ee.ExitCode()
153 if code < 0 { // killed by a signal: 128+signal like a shell254 if code < 0 { // killed by a signal: 128+signal like a shell
M internal/guest/guest_test.go +172 -22
@@ -4,10 +4,15 @@ package guest
44
55 import (
66 "context"
7+ "fmt"
78 "net"
9+ "os"
810 "path/filepath"
911 "reflect"
12+ "strconv"
1013 "strings"
14+ "sync"
15+ "sync/atomic"
1116 "syscall"
1217 "testing"
1318 "time"
@@ -45,7 +50,7 @@ func TestParseTarget(t *testing.T) {
4550 }
4651
4752 // host plays the host process on a unix socket and runs one session.
48-func host(t *testing.T, script string, ctxFn func(context.Context) context.Context) (out, errOut string, exit *execpb.Exit, hello *execpb.Hello, code int, execErr error) {
53+func host(t *testing.T, script string, cancelAfter time.Duration, tweak func(*ExecConfig)) (out, errOut string, exit *execpb.Exit, hello *execpb.Hello, code int, execErr error) {
4954 t.Helper()
5055 sock := filepath.Join(t.TempDir(), "s")
5156 ln, err := net.Listen("unix", sock)
@@ -59,7 +64,11 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
5964 }
6065 done := make(chan res, 1)
6166 go func() {
62- c, err := Exec(context.Background(), ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", Shell: "/bin/sh", RunTimeout: 5 * time.Second})
67+ cfg := ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", Shell: "/bin/sh", RunTimeout: 5 * time.Second}
68+ if tweak != nil {
69+ tweak(&cfg)
70+ }
71+ c, err := Exec(context.Background(), cfg)
6372 done <- res{c, err}
6473 }()
6574 conn, err := ln.Accept()
@@ -67,12 +76,13 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
6776 t.Fatal(err)
6877 }
6978 defer func() { _ = conn.Close() }()
70- ctx := context.Background()
71- if ctxFn != nil {
72- ctx = ctxFn(ctx)
79+ var cancel chan *execpb.Cancel
80+ if cancelAfter > 0 {
81+ cancel = make(chan *execpb.Cancel, 1)
82+ time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test"} })
7383 }
7484 var so, se strings.Builder
75- hello, exit, err = execstream.Drive(ctx, conn, &execpb.Run{Script: script, Env: map[string]string{"STUB_VAR": "from-host"}}, func(o *execpb.Output) {
85+ hello, exit, err = execstream.Drive(conn, &execpb.Run{Script: script, Env: map[string]string{"STUB_VAR": "from-host"}}, cancel, func(o *execpb.Output) {
7686 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
7787 se.Write(o.Data)
7888 } else {
@@ -87,7 +97,7 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
8797 }
8898
8999 func TestExecStreamsOutputAndExitCode(t *testing.T) {
90- out, errOut, exit, hello, code, err := host(t, `echo out-line; echo err-line >&2; echo "$STUB_VAR"; exit 3`, nil)
100+ out, errOut, exit, hello, code, err := host(t, `echo out-line; echo err-line >&2; echo "$STUB_VAR"; exit 3`, 0, nil)
91101 if err != nil {
92102 t.Fatal(err)
93103 }
@@ -105,7 +115,7 @@ func TestExecStreamsOutputAndExitCode(t *testing.T) {
105115 func TestExecDefaultScriptRunsDiagnostics(t *testing.T) {
106116 // The default script is the CI-1 stub: uname, id, os-release, docker info.
107117 // docker may be absent here: only the layout of the output is checked.
108- out, errOut, exit, _, _, _ := host(t, "", nil)
118+ out, errOut, exit, _, _, _ := host(t, "", 0, nil)
109119 if !strings.Contains(out, "Linux") || !strings.Contains(out, "uid=") {
110120 t.Fatalf("uname/id missing: out=%q err=%q", out, errOut)
111121 }
@@ -118,7 +128,7 @@ func TestExecDefaultScriptRunsDiagnostics(t *testing.T) {
118128 }
119129
120130 func TestExecLargeOutputIsChunkedAndComplete(t *testing.T) {
121- out, _, exit, _, _, _ := host(t, `head -c 300000 /dev/zero | tr '\0' 'x'`, nil)
131+ out, _, exit, _, _, _ := host(t, `head -c 300000 /dev/zero | tr '\0' 'x'`, 0, nil)
122132 if len(out) != 300000 || exit.Code != 0 {
123133 t.Fatalf("len=%d exit=%v", len(out), exit)
124134 }
@@ -126,11 +136,7 @@ func TestExecLargeOutputIsChunkedAndComplete(t *testing.T) {
126136
127137 func TestExecCancelSendsSIGINT(t *testing.T) {
128138 start := time.Now()
129- out, _, exit, _, _, _ := host(t, `trap 'echo got-int; exit 130' INT; echo ready; while :; do sleep 0.05; done`, func(ctx context.Context) context.Context {
130- c, cancel := context.WithCancel(ctx)
131- time.AfterFunc(500*time.Millisecond, cancel)
132- return c
133- })
139+ out, _, exit, _, _, _ := host(t, `trap 'echo got-int; exit 130' INT; echo ready; while :; do sleep 0.05; done`, 500*time.Millisecond, nil)
134140 if !strings.Contains(out, "got-int") || exit.Code != 130 {
135141 t.Fatalf("out=%q exit=%v", out, exit)
136142 }
@@ -139,8 +145,123 @@ func TestExecCancelSendsSIGINT(t *testing.T) {
139145 }
140146 }
141147
148+// sigRecorder replaces the signal function: it records (signal, time since
149+// start) and still delivers the signal.
150+type sigRecorder struct {
151+ mu sync.Mutex
152+ start time.Time
153+ got []string
154+ times []time.Duration
155+}
156+
157+func (r *sigRecorder) fn(pgid int, sig syscall.Signal) {
158+ r.mu.Lock()
159+ r.got = append(r.got, sig.String())
160+ r.times = append(r.times, time.Since(r.start))
161+ r.mu.Unlock()
162+ _ = syscall.Kill(-pgid, sig)
163+}
164+
165+// C1: a script that ignores SIGINT and SIGTERM is stopped by the escalation:
166+// SIGINT at once, SIGTERM after TermAfter, SIGKILL KillAfter later.
167+func TestExecCancelEscalatesToKill(t *testing.T) {
168+ rec := &sigRecorder{start: time.Now()}
169+ termAfter, killAfter := 400*time.Millisecond, 300*time.Millisecond
170+ _, _, exit, _, _, _ := host(t, `trap '' INT TERM; echo ready; while :; do sleep 0.05; done`, 300*time.Millisecond, func(c *ExecConfig) {
171+ c.TermAfter, c.KillAfter, c.signal = termAfter, killAfter, rec.fn
172+ })
173+ if exit.Code != 128+int32(syscall.SIGKILL) {
174+ t.Fatalf("a script ignoring INT and TERM must end by SIGKILL, exit=%v", exit)
175+ }
176+ rec.mu.Lock()
177+ defer rec.mu.Unlock()
178+ // SIGINT, SIGTERM, SIGKILL (from the escalation), then the final group kill
179+ // issued after the shell was reaped.
180+ if len(rec.got) < 3 || rec.got[0] != "interrupt" || rec.got[1] != "terminated" || rec.got[2] != "killed" {
181+ t.Fatalf("signals: %v", rec.got)
182+ }
183+ cancelAt := rec.times[0]
184+ if d := rec.times[1] - cancelAt; d < termAfter-50*time.Millisecond || d > termAfter+400*time.Millisecond {
185+ t.Errorf("SIGTERM %v after SIGINT, want about %v", d, termAfter)
186+ }
187+ if d := rec.times[2] - rec.times[1]; d < killAfter-50*time.Millisecond || d > killAfter+400*time.Millisecond {
188+ t.Errorf("SIGKILL %v after SIGTERM, want about %v", d, killAfter)
189+ }
190+}
191+
192+// C1 + C9: a script that exits on SIGINT stops the escalation: the timers are
193+// stopped when Wait returns, so no SIGTERM/SIGKILL follows, and nothing is
194+// signalled after the shell was reaped except the final group sweep.
195+func TestExecEscalationStopsWhenScriptExits(t *testing.T) {
196+ rec := &sigRecorder{start: time.Now()}
197+ _, _, exit, _, _, _ := host(t, `trap 'exit 130' INT; while :; do sleep 0.05; done`, 200*time.Millisecond, func(c *ExecConfig) {
198+ c.TermAfter, c.KillAfter, c.signal = 300*time.Millisecond, 300*time.Millisecond, rec.fn
199+ })
200+ if exit.Code != 130 {
201+ t.Fatalf("exit=%v", exit)
202+ }
203+ time.Sleep(1200 * time.Millisecond) // well past TermAfter+KillAfter
204+ rec.mu.Lock()
205+ defer rec.mu.Unlock()
206+ if len(rec.got) != 2 || rec.got[0] != "interrupt" || rec.got[1] != "killed" {
207+ t.Fatalf("want SIGINT then the final sweep only, got %v", rec.got)
208+ }
209+}
210+
211+// C9: after the shell was reaped the escalator never signals again, even if its
212+// timers had fired concurrently.
213+func TestEscalatorNeverSignalsAfterFinish(t *testing.T) {
214+ var n atomic.Int32
215+ e := &escalator{cfg: ExecConfig{TermAfter: time.Millisecond, KillAfter: time.Millisecond, signal: func(int, syscall.Signal) { n.Add(1) }}, pgid: 1 << 30}
216+ e.cancel()
217+ e.finish()
218+ after := n.Load()
219+ time.Sleep(50 * time.Millisecond)
220+ e.send(syscall.SIGTERM)
221+ e.cancel()
222+ if n.Load() != after {
223+ t.Fatalf("signalled %d times after finish", n.Load()-after)
224+ }
225+}
226+
227+// C2: a background child that keeps stdout open must not hang the executor: it
228+// reports the script's exit promptly and the child is killed.
229+func TestExecBackgroundChildDoesNotHoldTheSession(t *testing.T) {
230+ marker := filepath.Join(t.TempDir(), "child.pid")
231+ start := time.Now()
232+ out, _, exit, _, _, _ := host(t, `sleep 30 & echo $! > `+marker+`; echo parent-done`, 0, func(c *ExecConfig) { c.WaitDelay = 300 * time.Millisecond })
233+ if time.Since(start) > 5*time.Second {
234+ t.Fatalf("the session was held by a background child for %v", time.Since(start))
235+ }
236+ if exit == nil || exit.Code != 0 || !strings.Contains(out, "parent-done") {
237+ t.Fatalf("exit=%v out=%q", exit, out)
238+ }
239+ b, err := os.ReadFile(marker)
240+ if err != nil {
241+ t.Fatal(err)
242+ }
243+ pid, _ := strconv.Atoi(strings.TrimSpace(string(b)))
244+ deadline := time.Now().Add(3 * time.Second)
245+ for time.Now().Before(deadline) {
246+ if st, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); err != nil || strings.Contains(string(st), ") Z") {
247+ return
248+ }
249+ time.Sleep(50 * time.Millisecond)
250+ }
251+ t.Fatalf("background child %d survived the job", pid)
252+}
253+
254+// C2: output written by the script before the shell exits is complete even when
255+// the stream is large and a background child holds the pipe.
256+func TestExecOutputCompleteWithBackgroundChild(t *testing.T) {
257+ out, _, exit, _, _, _ := host(t, `sleep 30 & head -c 200000 /dev/zero | tr '\0' 'y'`, 0, func(c *ExecConfig) { c.WaitDelay = 300 * time.Millisecond })
258+ if exit.Code != 0 || len(out) != 200000 {
259+ t.Fatalf("len=%d exit=%v", len(out), exit)
260+ }
261+}
262+
142263 func TestExecKilledBySignalReports128PlusSignal(t *testing.T) {
143- _, _, exit, _, _, _ := host(t, `kill -TERM $$`, nil)
264+ _, _, exit, _, _, _ := host(t, `kill -TERM $$`, 0, nil)
144265 if exit.Code != 128+int32(syscall.SIGTERM) {
145266 t.Fatalf("exit=%v", exit)
146267 }
@@ -156,7 +277,7 @@ func TestExecMissingShell(t *testing.T) {
156277 go func() {
157278 c, _ := ln.Accept()
158279 defer func() { _ = c.Close() }()
159- _, exit, _ := execstream.Drive(context.Background(), c, &execpb.Run{Script: "true"}, nil)
280+ _, exit, _ := execstream.Drive(c, &execpb.Run{Script: "true"}, nil, nil)
160281 if exit == nil || exit.Code != 127 || exit.Error == "" {
161282 t.Errorf("want 127 with an error message, got %v", exit)
162283 }
@@ -233,14 +354,13 @@ func TestInitInContainerSkipsVMSetup(t *testing.T) {
233354 return
234355 }
235356 defer func() { _ = c.Close() }()
236- _, _, _ = execstream.Drive(context.Background(), c, &execpb.Run{Script: "exit 0"}, nil)
357+ _, _, _ = execstream.Drive(c, &execpb.Run{Script: "exit 0"}, nil, nil)
237358 }()
238359 code := Init(context.Background(), InitConfig{
239- Cmdline: "ermite.sandbox=vm ermite.connect=unix:" + sock,
240- Getpid: func() int { return 4242 },
241- ExecArgs: []string{"/bin/sh", "-c", "exit 5"},
242- DockerTimeout: time.Millisecond,
243- SkipDocker: true,
360+ Cmdline: "ermite.sandbox=vm ermite.connect=unix:" + sock,
361+ Getpid: func() int { return 4242 },
362+ ExecArgs: []string{"/bin/sh", "-c", "exit 5"},
363+ SkipDocker: true,
244364 })
245365 // ExecArgs is a stand-in for the executor: its exit code is Init's.
246366 if code != 5 {
@@ -259,3 +379,33 @@ func TestWriteResolvConfNeedsGateway(t *testing.T) {
259379 t.Fatal("empty gateway accepted")
260380 }
261381 }
382+
383+// C11: init forwards SIGTERM to the executor and finishes with the executor's
384+// own exit code, instead of dying with its child still running.
385+func TestInitForwardsSIGTERMToTheExecutor(t *testing.T) {
386+ sigs := make(chan os.Signal, 1)
387+ done := make(chan int, 1)
388+ marker := filepath.Join(t.TempDir(), "got-term")
389+ go func() {
390+ done <- Init(context.Background(), InitConfig{
391+ Cmdline: "ermite.sandbox=container ermite.connect=unix:/nonexistent",
392+ Getpid: func() int { return 4242 },
393+ SkipDocker: true,
394+ Signals: sigs,
395+ ExecArgs: []string{"/bin/sh", "-c", `trap 'echo term > ` + marker + `; exit 9' TERM; while :; do sleep 0.05; done`},
396+ })
397+ }()
398+ time.Sleep(500 * time.Millisecond)
399+ sigs <- syscall.SIGTERM
400+ select {
401+ case code := <-done:
402+ if code != 9 {
403+ t.Fatalf("init must return the executor's exit code, got %d", code)
404+ }
405+ case <-time.After(10 * time.Second):
406+ t.Fatal("init did not finish after SIGTERM")
407+ }
408+ if _, err := os.Stat(marker); err != nil {
409+ t.Fatal("the executor never received SIGTERM")
410+ }
411+}
@@ -4,10 +4,15 @@ package guest
4 4
5 import (5 import (
6 "context"6 "context"
7+ "fmt"
7 "net"8 "net"
9+ "os"
8 "path/filepath"10 "path/filepath"
9 "reflect"11 "reflect"
12+ "strconv"
10 "strings"13 "strings"
14+ "sync"
15+ "sync/atomic"
11 "syscall"16 "syscall"
12 "testing"17 "testing"
13 "time"18 "time"
@@ -45,7 +50,7 @@ func TestParseTarget(t *testing.T) {
45 }50 }
46 51
47 // host plays the host process on a unix socket and runs one session.52 // host plays the host process on a unix socket and runs one session.
48-func host(t *testing.T, script string, ctxFn func(context.Context) context.Context) (out, errOut string, exit *execpb.Exit, hello *execpb.Hello, code int, execErr error) {53+func host(t *testing.T, script string, cancelAfter time.Duration, tweak func(*ExecConfig)) (out, errOut string, exit *execpb.Exit, hello *execpb.Hello, code int, execErr error) {
49 t.Helper()54 t.Helper()
50 sock := filepath.Join(t.TempDir(), "s")55 sock := filepath.Join(t.TempDir(), "s")
51 ln, err := net.Listen("unix", sock)56 ln, err := net.Listen("unix", sock)
@@ -59,7 +64,11 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
59 }64 }
60 done := make(chan res, 1)65 done := make(chan res, 1)
61 go func() {66 go func() {
62- c, err := Exec(context.Background(), ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", Shell: "/bin/sh", RunTimeout: 5 * time.Second})67+ cfg := ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", Shell: "/bin/sh", RunTimeout: 5 * time.Second}
68+ if tweak != nil {
69+ tweak(&cfg)
70+ }
71+ c, err := Exec(context.Background(), cfg)
63 done <- res{c, err}72 done <- res{c, err}
64 }()73 }()
65 conn, err := ln.Accept()74 conn, err := ln.Accept()
@@ -67,12 +76,13 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
67 t.Fatal(err)76 t.Fatal(err)
68 }77 }
69 defer func() { _ = conn.Close() }()78 defer func() { _ = conn.Close() }()
70- ctx := context.Background()79+ var cancel chan *execpb.Cancel
71- if ctxFn != nil {80+ if cancelAfter > 0 {
72- ctx = ctxFn(ctx)81+ cancel = make(chan *execpb.Cancel, 1)
82+ time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test"} })
73 }83 }
74 var so, se strings.Builder84 var so, se strings.Builder
75- hello, exit, err = execstream.Drive(ctx, conn, &execpb.Run{Script: script, Env: map[string]string{"STUB_VAR": "from-host"}}, func(o *execpb.Output) {85+ hello, exit, err = execstream.Drive(conn, &execpb.Run{Script: script, Env: map[string]string{"STUB_VAR": "from-host"}}, cancel, func(o *execpb.Output) {
76 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {86 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
77 se.Write(o.Data)87 se.Write(o.Data)
78 } else {88 } else {
@@ -87,7 +97,7 @@ func host(t *testing.T, script string, ctxFn func(context.Context) context.Conte
87 }97 }
88 98
89 func TestExecStreamsOutputAndExitCode(t *testing.T) {99 func TestExecStreamsOutputAndExitCode(t *testing.T) {
90- out, errOut, exit, hello, code, err := host(t, `echo out-line; echo err-line >&2; echo "$STUB_VAR"; exit 3`, nil)100+ out, errOut, exit, hello, code, err := host(t, `echo out-line; echo err-line >&2; echo "$STUB_VAR"; exit 3`, 0, nil)
91 if err != nil {101 if err != nil {
92 t.Fatal(err)102 t.Fatal(err)
93 }103 }
@@ -105,7 +115,7 @@ func TestExecStreamsOutputAndExitCode(t *testing.T) {
105 func TestExecDefaultScriptRunsDiagnostics(t *testing.T) {115 func TestExecDefaultScriptRunsDiagnostics(t *testing.T) {
106 // The default script is the CI-1 stub: uname, id, os-release, docker info.116 // The default script is the CI-1 stub: uname, id, os-release, docker info.
107 // docker may be absent here: only the layout of the output is checked.117 // docker may be absent here: only the layout of the output is checked.
108- out, errOut, exit, _, _, _ := host(t, "", nil)118+ out, errOut, exit, _, _, _ := host(t, "", 0, nil)
109 if !strings.Contains(out, "Linux") || !strings.Contains(out, "uid=") {119 if !strings.Contains(out, "Linux") || !strings.Contains(out, "uid=") {
110 t.Fatalf("uname/id missing: out=%q err=%q", out, errOut)120 t.Fatalf("uname/id missing: out=%q err=%q", out, errOut)
111 }121 }
@@ -118,7 +128,7 @@ func TestExecDefaultScriptRunsDiagnostics(t *testing.T) {
118 }128 }
119 129
120 func TestExecLargeOutputIsChunkedAndComplete(t *testing.T) {130 func TestExecLargeOutputIsChunkedAndComplete(t *testing.T) {
121- out, _, exit, _, _, _ := host(t, `head -c 300000 /dev/zero | tr '\0' 'x'`, nil)131+ out, _, exit, _, _, _ := host(t, `head -c 300000 /dev/zero | tr '\0' 'x'`, 0, nil)
122 if len(out) != 300000 || exit.Code != 0 {132 if len(out) != 300000 || exit.Code != 0 {
123 t.Fatalf("len=%d exit=%v", len(out), exit)133 t.Fatalf("len=%d exit=%v", len(out), exit)
124 }134 }
@@ -126,11 +136,7 @@ func TestExecLargeOutputIsChunkedAndComplete(t *testing.T) {
126 136
127 func TestExecCancelSendsSIGINT(t *testing.T) {137 func TestExecCancelSendsSIGINT(t *testing.T) {
128 start := time.Now()138 start := time.Now()
129- out, _, exit, _, _, _ := host(t, `trap 'echo got-int; exit 130' INT; echo ready; while :; do sleep 0.05; done`, func(ctx context.Context) context.Context {139+ out, _, exit, _, _, _ := host(t, `trap 'echo got-int; exit 130' INT; echo ready; while :; do sleep 0.05; done`, 500*time.Millisecond, nil)
130- c, cancel := context.WithCancel(ctx)
131- time.AfterFunc(500*time.Millisecond, cancel)
132- return c
133- })
134 if !strings.Contains(out, "got-int") || exit.Code != 130 {140 if !strings.Contains(out, "got-int") || exit.Code != 130 {
135 t.Fatalf("out=%q exit=%v", out, exit)141 t.Fatalf("out=%q exit=%v", out, exit)
136 }142 }
@@ -139,8 +145,123 @@ func TestExecCancelSendsSIGINT(t *testing.T) {
139 }145 }
140 }146 }
141 147
148+// sigRecorder replaces the signal function: it records (signal, time since
149+// start) and still delivers the signal.
150+type sigRecorder struct {
151+ mu sync.Mutex
152+ start time.Time
153+ got []string
154+ times []time.Duration
155+}
156+
157+func (r *sigRecorder) fn(pgid int, sig syscall.Signal) {
158+ r.mu.Lock()
159+ r.got = append(r.got, sig.String())
160+ r.times = append(r.times, time.Since(r.start))
161+ r.mu.Unlock()
162+ _ = syscall.Kill(-pgid, sig)
163+}
164+
165+// C1: a script that ignores SIGINT and SIGTERM is stopped by the escalation:
166+// SIGINT at once, SIGTERM after TermAfter, SIGKILL KillAfter later.
167+func TestExecCancelEscalatesToKill(t *testing.T) {
168+ rec := &sigRecorder{start: time.Now()}
169+ termAfter, killAfter := 400*time.Millisecond, 300*time.Millisecond
170+ _, _, exit, _, _, _ := host(t, `trap '' INT TERM; echo ready; while :; do sleep 0.05; done`, 300*time.Millisecond, func(c *ExecConfig) {
171+ c.TermAfter, c.KillAfter, c.signal = termAfter, killAfter, rec.fn
172+ })
173+ if exit.Code != 128+int32(syscall.SIGKILL) {
174+ t.Fatalf("a script ignoring INT and TERM must end by SIGKILL, exit=%v", exit)
175+ }
176+ rec.mu.Lock()
177+ defer rec.mu.Unlock()
178+ // SIGINT, SIGTERM, SIGKILL (from the escalation), then the final group kill
179+ // issued after the shell was reaped.
180+ if len(rec.got) < 3 || rec.got[0] != "interrupt" || rec.got[1] != "terminated" || rec.got[2] != "killed" {
181+ t.Fatalf("signals: %v", rec.got)
182+ }
183+ cancelAt := rec.times[0]
184+ if d := rec.times[1] - cancelAt; d < termAfter-50*time.Millisecond || d > termAfter+400*time.Millisecond {
185+ t.Errorf("SIGTERM %v after SIGINT, want about %v", d, termAfter)
186+ }
187+ if d := rec.times[2] - rec.times[1]; d < killAfter-50*time.Millisecond || d > killAfter+400*time.Millisecond {
188+ t.Errorf("SIGKILL %v after SIGTERM, want about %v", d, killAfter)
189+ }
190+}
191+
192+// C1 + C9: a script that exits on SIGINT stops the escalation: the timers are
193+// stopped when Wait returns, so no SIGTERM/SIGKILL follows, and nothing is
194+// signalled after the shell was reaped except the final group sweep.
195+func TestExecEscalationStopsWhenScriptExits(t *testing.T) {
196+ rec := &sigRecorder{start: time.Now()}
197+ _, _, exit, _, _, _ := host(t, `trap 'exit 130' INT; while :; do sleep 0.05; done`, 200*time.Millisecond, func(c *ExecConfig) {
198+ c.TermAfter, c.KillAfter, c.signal = 300*time.Millisecond, 300*time.Millisecond, rec.fn
199+ })
200+ if exit.Code != 130 {
201+ t.Fatalf("exit=%v", exit)
202+ }
203+ time.Sleep(1200 * time.Millisecond) // well past TermAfter+KillAfter
204+ rec.mu.Lock()
205+ defer rec.mu.Unlock()
206+ if len(rec.got) != 2 || rec.got[0] != "interrupt" || rec.got[1] != "killed" {
207+ t.Fatalf("want SIGINT then the final sweep only, got %v", rec.got)
208+ }
209+}
210+
211+// C9: after the shell was reaped the escalator never signals again, even if its
212+// timers had fired concurrently.
213+func TestEscalatorNeverSignalsAfterFinish(t *testing.T) {
214+ var n atomic.Int32
215+ e := &escalator{cfg: ExecConfig{TermAfter: time.Millisecond, KillAfter: time.Millisecond, signal: func(int, syscall.Signal) { n.Add(1) }}, pgid: 1 << 30}
216+ e.cancel()
217+ e.finish()
218+ after := n.Load()
219+ time.Sleep(50 * time.Millisecond)
220+ e.send(syscall.SIGTERM)
221+ e.cancel()
222+ if n.Load() != after {
223+ t.Fatalf("signalled %d times after finish", n.Load()-after)
224+ }
225+}
226+
227+// C2: a background child that keeps stdout open must not hang the executor: it
228+// reports the script's exit promptly and the child is killed.
229+func TestExecBackgroundChildDoesNotHoldTheSession(t *testing.T) {
230+ marker := filepath.Join(t.TempDir(), "child.pid")
231+ start := time.Now()
232+ out, _, exit, _, _, _ := host(t, `sleep 30 & echo $! > `+marker+`; echo parent-done`, 0, func(c *ExecConfig) { c.WaitDelay = 300 * time.Millisecond })
233+ if time.Since(start) > 5*time.Second {
234+ t.Fatalf("the session was held by a background child for %v", time.Since(start))
235+ }
236+ if exit == nil || exit.Code != 0 || !strings.Contains(out, "parent-done") {
237+ t.Fatalf("exit=%v out=%q", exit, out)
238+ }
239+ b, err := os.ReadFile(marker)
240+ if err != nil {
241+ t.Fatal(err)
242+ }
243+ pid, _ := strconv.Atoi(strings.TrimSpace(string(b)))
244+ deadline := time.Now().Add(3 * time.Second)
245+ for time.Now().Before(deadline) {
246+ if st, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); err != nil || strings.Contains(string(st), ") Z") {
247+ return
248+ }
249+ time.Sleep(50 * time.Millisecond)
250+ }
251+ t.Fatalf("background child %d survived the job", pid)
252+}
253+
254+// C2: output written by the script before the shell exits is complete even when
255+// the stream is large and a background child holds the pipe.
256+func TestExecOutputCompleteWithBackgroundChild(t *testing.T) {
257+ out, _, exit, _, _, _ := host(t, `sleep 30 & head -c 200000 /dev/zero | tr '\0' 'y'`, 0, func(c *ExecConfig) { c.WaitDelay = 300 * time.Millisecond })
258+ if exit.Code != 0 || len(out) != 200000 {
259+ t.Fatalf("len=%d exit=%v", len(out), exit)
260+ }
261+}
262+
142 func TestExecKilledBySignalReports128PlusSignal(t *testing.T) {263 func TestExecKilledBySignalReports128PlusSignal(t *testing.T) {
143- _, _, exit, _, _, _ := host(t, `kill -TERM $$`, nil)264+ _, _, exit, _, _, _ := host(t, `kill -TERM $$`, 0, nil)
144 if exit.Code != 128+int32(syscall.SIGTERM) {265 if exit.Code != 128+int32(syscall.SIGTERM) {
145 t.Fatalf("exit=%v", exit)266 t.Fatalf("exit=%v", exit)
146 }267 }
@@ -156,7 +277,7 @@ func TestExecMissingShell(t *testing.T) {
156 go func() {277 go func() {
157 c, _ := ln.Accept()278 c, _ := ln.Accept()
158 defer func() { _ = c.Close() }()279 defer func() { _ = c.Close() }()
159- _, exit, _ := execstream.Drive(context.Background(), c, &execpb.Run{Script: "true"}, nil)280+ _, exit, _ := execstream.Drive(c, &execpb.Run{Script: "true"}, nil, nil)
160 if exit == nil || exit.Code != 127 || exit.Error == "" {281 if exit == nil || exit.Code != 127 || exit.Error == "" {
161 t.Errorf("want 127 with an error message, got %v", exit)282 t.Errorf("want 127 with an error message, got %v", exit)
162 }283 }
@@ -233,14 +354,13 @@ func TestInitInContainerSkipsVMSetup(t *testing.T) {
233 return354 return
234 }355 }
235 defer func() { _ = c.Close() }()356 defer func() { _ = c.Close() }()
236- _, _, _ = execstream.Drive(context.Background(), c, &execpb.Run{Script: "exit 0"}, nil)357+ _, _, _ = execstream.Drive(c, &execpb.Run{Script: "exit 0"}, nil, nil)
237 }()358 }()
238 code := Init(context.Background(), InitConfig{359 code := Init(context.Background(), InitConfig{
239- Cmdline: "ermite.sandbox=vm ermite.connect=unix:" + sock,360+ Cmdline: "ermite.sandbox=vm ermite.connect=unix:" + sock,
240- Getpid: func() int { return 4242 },361+ Getpid: func() int { return 4242 },
241- ExecArgs: []string{"/bin/sh", "-c", "exit 5"},362+ ExecArgs: []string{"/bin/sh", "-c", "exit 5"},
242- DockerTimeout: time.Millisecond,363+ SkipDocker: true,
243- SkipDocker: true,
244 })364 })
245 // ExecArgs is a stand-in for the executor: its exit code is Init's.365 // ExecArgs is a stand-in for the executor: its exit code is Init's.
246 if code != 5 {366 if code != 5 {
@@ -259,3 +379,33 @@ func TestWriteResolvConfNeedsGateway(t *testing.T) {
259 t.Fatal("empty gateway accepted")379 t.Fatal("empty gateway accepted")
260 }380 }
261 }381 }
382+
383+// C11: init forwards SIGTERM to the executor and finishes with the executor's
384+// own exit code, instead of dying with its child still running.
385+func TestInitForwardsSIGTERMToTheExecutor(t *testing.T) {
386+ sigs := make(chan os.Signal, 1)
387+ done := make(chan int, 1)
388+ marker := filepath.Join(t.TempDir(), "got-term")
389+ go func() {
390+ done <- Init(context.Background(), InitConfig{
391+ Cmdline: "ermite.sandbox=container ermite.connect=unix:/nonexistent",
392+ Getpid: func() int { return 4242 },
393+ SkipDocker: true,
394+ Signals: sigs,
395+ ExecArgs: []string{"/bin/sh", "-c", `trap 'echo term > ` + marker + `; exit 9' TERM; while :; do sleep 0.05; done`},
396+ })
397+ }()
398+ time.Sleep(500 * time.Millisecond)
399+ sigs <- syscall.SIGTERM
400+ select {
401+ case code := <-done:
402+ if code != 9 {
403+ t.Fatalf("init must return the executor's exit code, got %d", code)
404+ }
405+ case <-time.After(10 * time.Second):
406+ t.Fatal("init did not finish after SIGTERM")
407+ }
408+ if _, err := os.Stat(marker); err != nil {
409+ t.Fatal("the executor never received SIGTERM")
410+ }
411+}
M internal/guest/init.go +29 -21
@@ -6,6 +6,7 @@ import (
66 "context"
77 "fmt"
88 "os"
9+ "os/signal"
910 "strings"
1011 "syscall"
1112 "time"
@@ -24,8 +25,9 @@ type InitConfig struct {
2425 Cmdline string
2526 // Getpid reports the process id; Init only acts as a VM's PID 1 when it is 1.
2627 Getpid func() int
27- // DockerTimeout bounds the wait for the Docker socket (default 60 s).
28- DockerTimeout time.Duration
28+ // Signals delivers the termination signals to forward to the executor;
29+ // nil means "subscribe to SIGTERM and SIGINT". Tests inject their own.
30+ Signals <-chan os.Signal
2931 // SkipDocker does not start dockerd (tests, images without Docker).
3032 SkipDocker bool
3133 // ExecArgs is the executor command line; default
@@ -51,9 +53,6 @@ func Init(ctx context.Context, cfg InitConfig) int {
5153 if cfg.Getpid == nil {
5254 cfg.Getpid = os.Getpid
5355 }
54- if cfg.DockerTimeout == 0 {
55- cfg.DockerTimeout = 60 * time.Second
56- }
5756 if cfg.ExecArgs == nil {
5857 cfg.ExecArgs = []string{sandbox.GuestExecDir + "/" + sandbox.GuestExecName, "exec"}
5958 }
@@ -134,11 +133,32 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
134133 log.Error().Err(err).Msg("init: start executor")
135134 return 1
136135 }
137- // Docker is awaited in the background: the executor starts right away and
138- // the stub script's `docker info` waits for the socket itself.
139- if dockerPid != 0 {
140- go waitSocket(ctx, "/var/run/docker.sock", cfg.DockerTimeout)
136+ // SIGTERM/SIGINT (docker stop, or the host stopping the VM) go to the
137+ // executor, which cancels its script with the usual escalation and reports
138+ // its Exit; init then finishes normally instead of dying with children.
139+ sigs := cfg.Signals
140+ if sigs == nil {
141+ ch := make(chan os.Signal, 4)
142+ signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT)
143+ defer signal.Stop(ch)
144+ sigs = ch
141145 }
146+ stopFwd := make(chan struct{})
147+ defer close(stopFwd)
148+ go func() {
149+ for {
150+ select {
151+ case sig := <-sigs:
152+ if s, ok := sig.(syscall.Signal); ok {
153+ log.Info().Str("signal", s.String()).Msg("init: forwarding the signal to the executor")
154+ _ = syscall.Kill(-execPid, s) // the executor leads its own process group
155+ _ = syscall.Kill(execPid, s)
156+ }
157+ case <-stopFwd:
158+ return
159+ }
160+ }
161+ }()
142162 code := reapUntil(execPid)
143163 if dockerPid != 0 {
144164 _ = syscall.Kill(dockerPid, syscall.SIGTERM)
@@ -154,18 +174,6 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
154174 return code
155175 }
156176
157-func waitSocket(ctx context.Context, path string, timeout time.Duration) {
158- log := zerowrap.FromCtx(ctx)
159- deadline := time.Now().Add(timeout)
160- for time.Now().Before(deadline) {
161- if _, err := os.Stat(path); err == nil {
162- return
163- }
164- time.Sleep(200 * time.Millisecond)
165- }
166- log.Warn().Str("socket", path).Msg("init: docker socket did not appear")
167-}
168-
169177 // spawn starts argv[0] as a child (own process group) without os/exec, so that
170178 // reapUntil is the only waiter and orphans can be reaped safely.
171179 func spawn(argv, env []string) (int, error) {
@@ -6,6 +6,7 @@ import (
6 "context"6 "context"
7 "fmt"7 "fmt"
8 "os"8 "os"
9+ "os/signal"
9 "strings"10 "strings"
10 "syscall"11 "syscall"
11 "time"12 "time"
@@ -24,8 +25,9 @@ type InitConfig struct {
24 Cmdline string25 Cmdline string
25 // Getpid reports the process id; Init only acts as a VM's PID 1 when it is 1.26 // Getpid reports the process id; Init only acts as a VM's PID 1 when it is 1.
26 Getpid func() int27 Getpid func() int
27- // DockerTimeout bounds the wait for the Docker socket (default 60 s).28+ // Signals delivers the termination signals to forward to the executor;
28- DockerTimeout time.Duration29+ // nil means "subscribe to SIGTERM and SIGINT". Tests inject their own.
30+ Signals <-chan os.Signal
29 // SkipDocker does not start dockerd (tests, images without Docker).31 // SkipDocker does not start dockerd (tests, images without Docker).
30 SkipDocker bool32 SkipDocker bool
31 // ExecArgs is the executor command line; default33 // ExecArgs is the executor command line; default
@@ -51,9 +53,6 @@ func Init(ctx context.Context, cfg InitConfig) int {
51 if cfg.Getpid == nil {53 if cfg.Getpid == nil {
52 cfg.Getpid = os.Getpid54 cfg.Getpid = os.Getpid
53 }55 }
54- if cfg.DockerTimeout == 0 {
55- cfg.DockerTimeout = 60 * time.Second
56- }
57 if cfg.ExecArgs == nil {56 if cfg.ExecArgs == nil {
58 cfg.ExecArgs = []string{sandbox.GuestExecDir + "/" + sandbox.GuestExecName, "exec"}57 cfg.ExecArgs = []string{sandbox.GuestExecDir + "/" + sandbox.GuestExecName, "exec"}
59 }58 }
@@ -134,11 +133,32 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
134 log.Error().Err(err).Msg("init: start executor")133 log.Error().Err(err).Msg("init: start executor")
135 return 1134 return 1
136 }135 }
137- // Docker is awaited in the background: the executor starts right away and136+ // SIGTERM/SIGINT (docker stop, or the host stopping the VM) go to the
138- // the stub script's `docker info` waits for the socket itself.137+ // executor, which cancels its script with the usual escalation and reports
139- if dockerPid != 0 {138+ // its Exit; init then finishes normally instead of dying with children.
140- go waitSocket(ctx, "/var/run/docker.sock", cfg.DockerTimeout)139+ sigs := cfg.Signals
140+ if sigs == nil {
141+ ch := make(chan os.Signal, 4)
142+ signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT)
143+ defer signal.Stop(ch)
144+ sigs = ch
141 }145 }
146+ stopFwd := make(chan struct{})
147+ defer close(stopFwd)
148+ go func() {
149+ for {
150+ select {
151+ case sig := <-sigs:
152+ if s, ok := sig.(syscall.Signal); ok {
153+ log.Info().Str("signal", s.String()).Msg("init: forwarding the signal to the executor")
154+ _ = syscall.Kill(-execPid, s) // the executor leads its own process group
155+ _ = syscall.Kill(execPid, s)
156+ }
157+ case <-stopFwd:
158+ return
159+ }
160+ }
161+ }()
142 code := reapUntil(execPid)162 code := reapUntil(execPid)
143 if dockerPid != 0 {163 if dockerPid != 0 {
144 _ = syscall.Kill(dockerPid, syscall.SIGTERM)164 _ = syscall.Kill(dockerPid, syscall.SIGTERM)
@@ -154,18 +174,6 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
154 return code174 return code
155 }175 }
156 176
157-func waitSocket(ctx context.Context, path string, timeout time.Duration) {
158- log := zerowrap.FromCtx(ctx)
159- deadline := time.Now().Add(timeout)
160- for time.Now().Before(deadline) {
161- if _, err := os.Stat(path); err == nil {
162- return
163- }
164- time.Sleep(200 * time.Millisecond)
165- }
166- log.Warn().Str("socket", path).Msg("init: docker socket did not appear")
167-}
168-
169 // spawn starts argv[0] as a child (own process group) without os/exec, so that177 // spawn starts argv[0] as a child (own process group) without os/exec, so that
170 // reapUntil is the only waiter and orphans can be reaped safely.178 // reapUntil is the only waiter and orphans can be reaped safely.
171 func spawn(argv, env []string) (int, error) {179 func spawn(argv, env []string) (int, error) {
M internal/sandbox/container/container_test.go +1 -1
@@ -242,7 +242,7 @@ func (e *cenv) leftovers() []string {
242242 func stub(t *testing.T, conn sandbox.Conn) (string, int) {
243243 t.Helper()
244244 var out strings.Builder
245- _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })
245+ _, exit, err := execstream.Drive(conn, &execpb.Run{}, nil, func(o *execpb.Output) { out.Write(o.Data) })
246246 if err != nil {
247247 t.Fatal(err)
248248 }
@@ -242,7 +242,7 @@ func (e *cenv) leftovers() []string {
242 func stub(t *testing.T, conn sandbox.Conn) (string, int) {242 func stub(t *testing.T, conn sandbox.Conn) (string, int) {
243 t.Helper()243 t.Helper()
244 var out strings.Builder244 var out strings.Builder
245- _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })245+ _, exit, err := execstream.Drive(conn, &execpb.Run{}, nil, func(o *execpb.Output) { out.Write(o.Data) })
246 if err != nil {246 if err != nil {
247 t.Fatal(err)247 t.Fatal(err)
248 }248 }
M internal/sandbox/container/docker_api_test.go +1 -1
@@ -73,7 +73,7 @@ func TestDockerAPIWiring(t *testing.T) {
7373 t.Fatalf("job %d: %v", i, err)
7474 }
7575 var out strings.Builder
76- hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, func(o *execpb.Output) { out.Write(o.Data) })
76+ hello, exit, err := execstream.Drive(conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, nil, func(o *execpb.Output) { out.Write(o.Data) })
7777 if err != nil {
7878 t.Fatal(err)
7979 }
@@ -73,7 +73,7 @@ func TestDockerAPIWiring(t *testing.T) {
73 t.Fatalf("job %d: %v", i, err)73 t.Fatalf("job %d: %v", i, err)
74 }74 }
75 var out strings.Builder75 var out strings.Builder
76- hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, func(o *execpb.Output) { out.Write(o.Data) })76+ hello, exit, err := execstream.Drive(conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, nil, func(o *execpb.Output) { out.Write(o.Data) })
77 if err != nil {77 if err != nil {
78 t.Fatal(err)78 t.Fatal(err)
79 }79 }
M internal/sandbox/integration/integration_test.go +1 -1
@@ -125,7 +125,7 @@ type result struct {
125125
126126 func runJob(ctx context.Context, conn sandbox.Conn, script string) (result, error) {
127127 var so, se strings.Builder
128- hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: script}, func(o *execpb.Output) {
128+ hello, exit, err := execstream.Drive(conn, &execpb.Run{Script: script}, nil, func(o *execpb.Output) {
129129 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
130130 se.Write(o.Data)
131131 } else {
@@ -125,7 +125,7 @@ type result struct {
125 125
126 func runJob(ctx context.Context, conn sandbox.Conn, script string) (result, error) {126 func runJob(ctx context.Context, conn sandbox.Conn, script string) (result, error) {
127 var so, se strings.Builder127 var so, se strings.Builder
128- hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: script}, func(o *execpb.Output) {128+ hello, exit, err := execstream.Drive(conn, &execpb.Run{Script: script}, nil, func(o *execpb.Output) {
129 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {129 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
130 se.Write(o.Data)130 se.Write(o.Data)
131 } else {131 } else {
M internal/sandbox/vm/vm_test.go +1 -1
@@ -196,7 +196,7 @@ func (e *env) leftovers(t *testing.T) []string {
196196 func runStub(t *testing.T, conn sandbox.Conn) (string, int) {
197197 t.Helper()
198198 var out strings.Builder
199- _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })
199+ _, exit, err := execstream.Drive(conn, &execpb.Run{}, nil, func(o *execpb.Output) { out.Write(o.Data) })
200200 if err != nil {
201201 t.Fatal(err)
202202 }
@@ -196,7 +196,7 @@ func (e *env) leftovers(t *testing.T) []string {
196 func runStub(t *testing.T, conn sandbox.Conn) (string, int) {196 func runStub(t *testing.T, conn sandbox.Conn) (string, int) {
197 t.Helper()197 t.Helper()
198 var out strings.Builder198 var out strings.Builder
199- _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })199+ _, exit, err := execstream.Drive(conn, &execpb.Run{}, nil, func(o *execpb.Output) { out.Write(o.Data) })
200 if err != nil {200 if err != nil {
201 t.Fatal(err)201 t.Fatal(err)
202 }202 }
M sandbox_check.go +5 -1
@@ -55,7 +55,11 @@ func sandboxCheck(ctx context.Context, args []string) int {
5555 fmt.Fprintln(os.Stderr, "ermite-runner:", err)
5656 return 1
5757 }
58- _, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: strings.Join(args, " ")}, func(o *execpb.Output) {
58+ // Ctrl-C / SIGTERM cancel the job: the executor escalates by itself.
59+ cancel := make(chan *execpb.Cancel, 1)
60+ stop := context.AfterFunc(ctx, func() { cancel <- &execpb.Cancel{Reason: "interrupted"} })
61+ defer stop()
62+ _, exit, err := execstream.Drive(conn, &execpb.Run{Script: strings.Join(args, " ")}, cancel, func(o *execpb.Output) {
5963 f := os.Stdout
6064 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
6165 f = os.Stderr
@@ -55,7 +55,11 @@ func sandboxCheck(ctx context.Context, args []string) int {
55 fmt.Fprintln(os.Stderr, "ermite-runner:", err)55 fmt.Fprintln(os.Stderr, "ermite-runner:", err)
56 return 156 return 1
57 }57 }
58- _, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: strings.Join(args, " ")}, func(o *execpb.Output) {58+ // Ctrl-C / SIGTERM cancel the job: the executor escalates by itself.
59+ cancel := make(chan *execpb.Cancel, 1)
60+ stop := context.AfterFunc(ctx, func() { cancel <- &execpb.Cancel{Reason: "interrupted"} })
61+ defer stop()
62+ _, exit, err := execstream.Drive(conn, &execpb.Run{Script: strings.Join(args, " ")}, cancel, func(o *execpb.Output) {
59 f := os.Stdout63 f := os.Stdout
60 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {64 if o.Stream == execpb.StreamKind_STREAM_KIND_STDERR {
61 f = os.Stderr65 f = os.Stderr

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close