diff --git a/pkg/sandbox/e2b/gateway.go b/pkg/sandbox/e2b/gateway.go index d6d0c07d7..ec303ca85 100644 --- a/pkg/sandbox/e2b/gateway.go +++ b/pkg/sandbox/e2b/gateway.go @@ -246,11 +246,78 @@ func (s *Server) wakeForTraffic(r *http.Request, sandboxID string) (*pb.GetSessi return nil, false, connectError("unavailable", "auto-resume failed: "+rerr.Error()) } s.registry.markPaused(sandboxID, false) - return sess, true, nil + // Resume re-creates the pod asynchronously, so the session view we + // already hold still reports the pre-pause state: PodIp is empty + // because the old pod is gone and the new one has not been assigned an + // address yet. Callers address sandboxd by pod IP, so returning that + // stale view made every post-resume call fail with + // "session ... has no pod IP" (E2B watch_dir, files.stat, ...). + // + // Poll until the new pod reports an address. Bounded: a resume that + // never schedules a pod is an infrastructure failure, and blocking the + // request forever is worse than reporting it. + fresh, perr := s.waitForPodIP(r, sandboxID) + if perr != nil { + return nil, false, connectError("unavailable", "sandbox resumed but has no pod IP: "+perr.Error()) + } + return fresh, true, nil + } + // Running: normally return the live view. But a sandbox that an explicit + // resume() just brought back is already marked running while its pod is + // still being created, so its PodIp is momentarily empty. Callers address + // sandboxd by pod IP, so wait briefly rather than failing the request — + // otherwise the first files/process call after resume() reports + // "has no pod IP". + if st == stateRunning && sess.GetPodIp() == "" { + if fresh, perr := s.waitForPodIP(r, sandboxID); perr == nil { + return fresh, false, nil + } } return sess, false, nil } +// waitForPodIP polls the session until the scheduler has assigned the +// (re)created pod an address, returning the fresh session view. +// +// Poll rather than a single read because resume is asynchronous: the pod is +// claimed and scheduled after ResumeSession returns, and the pod IP cache in +// the sandboxd client is deliberately short-lived (e2bPodIPTTL) so it cannot +// paper over the gap. +func (s *Server) waitForPodIP(r *http.Request, sandboxID string) (*pb.GetSessionResponse, error) { + const ( + // podIPWaitTimeout bounds the whole wait. Warm-pool claims normally + // land in well under a second; a cold pod needs scheduling time. + podIPWaitTimeout = 30 * time.Second + podIPWaitPoll = 200 * time.Millisecond + ) + deadline := time.Now().Add(podIPWaitTimeout) + for { + sess, err := s.gw.GetSession(r.Context(), &pb.GetSessionRequest{SessionId: sandboxID}) + if err == nil && sess.GetPodIp() != "" { + // The pod changed, so any cached address for this session is stale. + s.invalidatePodIP(sandboxID) + return sess, nil + } + if !time.Now().Before(deadline) { + if err != nil { + return nil, err + } + return nil, fmt.Errorf("pod IP not assigned after %s", podIPWaitTimeout) + } + time.Sleep(podIPWaitPoll) + } +} + +// invalidatePodIP drops any cached sandboxd address for a session. Pause, +// resume and pod replacement all change the address, so the short pod-IP cache +// must be dropped at those boundaries or it keeps answering with an IP that no +// longer exists. +func (s *Server) invalidatePodIP(sandboxID string) { + if s.sandboxd != nil { + s.sandboxd.invalidatePodIP(sandboxID) + } +} + // unixNow returns the current unix time in seconds. func unixNow() int64 { return time.Now().Unix() } diff --git a/pkg/sandbox/e2b/gateway_podip_test.go b/pkg/sandbox/e2b/gateway_podip_test.go new file mode 100644 index 000000000..3e734923e --- /dev/null +++ b/pkg/sandbox/e2b/gateway_podip_test.go @@ -0,0 +1,158 @@ +package e2b + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + pb "github.com/xiaods/k8e/pkg/sandboxmatrix/grpc/pb/sandbox/v1" +) + +// asyncPodIPGateway models the real resume path: ResumeSession marks the +// session Active but the pod IP is only assigned a few polls later, exactly as +// the scheduler behaves. A single GetSession immediately after resume therefore +// reports an empty PodIp — which is what made the E2B layer answer +// "session ... has no pod IP". +type asyncPodIPGateway struct { + *fakeGateway + mu sync.Mutex + gets int + assignAt int // GetSession call number at/after which PodIp appears +} + +func (g *asyncPodIPGateway) GetSession(ctx context.Context, req *pb.GetSessionRequest) (*pb.GetSessionResponse, error) { + g.mu.Lock() + g.gets++ + n := g.gets + assign := n >= g.assignAt + g.mu.Unlock() + + sess, err := g.fakeGateway.GetSession(ctx, req) + if err != nil { + return nil, err + } + if assign && sess.GetPodIp() == "" { + // Simulate the scheduler finally assigning an address. + g.fakeGateway.mu.Lock() + g.fakeGateway.sessions[req.SessionId].PodIp = "10.42.0.99" + g.fakeGateway.mu.Unlock() + return g.fakeGateway.GetSession(ctx, req) + } + return sess, nil +} + +// PauseSession drops the pod address, mirroring the real controller: a paused +// sandbox has no running pod, so status.podIP is cleared until it is resumed. +// Without this the fake would leave a stale address behind and the pod-IP race +// would never be exercised. +func (g *asyncPodIPGateway) PauseSession(ctx context.Context, req *pb.PauseSessionRequest) (*pb.PauseSessionResponse, error) { + resp, err := g.fakeGateway.PauseSession(ctx, req) + if err != nil { + return nil, err + } + g.fakeGateway.mu.Lock() + if s, ok := g.fakeGateway.sessions[req.SessionId]; ok { + s.PodIp = "" + } + g.fakeGateway.mu.Unlock() + return resp, nil +} + +// listDir issues an authenticated envd ListDir, which is the request that +// triggers wakeForTraffic's auto-resume and then must reach sandboxd by pod IP. +func listDir(t *testing.T, ts *httptest.Server, s *Server, id string) { + t.Helper() + req, err := http.NewRequest(http.MethodPost, + ts.URL+"/e2b/envd/filesystem.Filesystem/ListDir", + strings.NewReader(`{"path":"/workspace","depth":1}`)) + if err != nil { + t.Fatal(err) + } + for k, v := range envdHeaders(t, s, id) { + req.Header.Set(k, v) + } + resp, err := ts.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("ListDir after resume: %d %s", resp.StatusCode, readBody(t, resp)) + } +} + +// TestWakeForTrafficWaitsForPodIPAfterResume is the regression test for the E2B +// bug where the first call after a pause failed with "session has no pod IP". +func TestWakeForTrafficWaitsForPodIPAfterResume(t *testing.T) { + base := newFakeGateway() + gw := &asyncPodIPGateway{fakeGateway: base, assignAt: 6} + stub := newSandboxdStub() + s, ts := testServerWithSandboxd(t, gw, stub) + defer ts.Close() + + id := createSandboxID(t, ts) + + // Seed the workspace so ListDir has something real to stat. + gw.mu.Lock() + gw.files["/workspace/probe.txt"] = "x" + gw.mu.Unlock() + stub.mu.Lock() + stub.dirs["/workspace/"] = true + stub.files["/workspace/probe.txt"] = "x" + stub.mu.Unlock() + + // Pause: the pod address goes away and the phase becomes Paused. + if resp := controlReq(t, ts, "POST", "/sandboxes/"+id+"/pause", nil); resp.StatusCode != 204 { + t.Fatalf("pause: %d %s", resp.StatusCode, readBody(t, resp)) + } + + // The call must auto-resume AND resolve to a usable pod IP. Before the fix + // it returned the pre-resume view whose PodIp was empty, and the sandboxd + // client answered "session ... has no pod IP". + listDir(t, ts, s, id) + + if got := len(base.resumed); got != 1 { + t.Fatalf("expected exactly one auto-resume, got %d", got) + } + if base.sessions[id].GetPodIp() == "" { + t.Fatal("auto-resume must leave the session with an assigned pod IP") + } +} + +// TestWakeForTrafficRunningWithoutPodIPWaits covers the explicit-resume case: +// the phase is already running while the pod is still being created. +func TestWakeForTrafficRunningWithoutPodIPWaits(t *testing.T) { + base := newFakeGateway() + gw := &asyncPodIPGateway{fakeGateway: base, assignAt: 3} + stub := newSandboxdStub() + s, ts := testServerWithSandboxd(t, gw, stub) + defer ts.Close() + + id := createSandboxID(t, ts) + + gw.mu.Lock() + gw.files["/workspace/probe.txt"] = "x" + gw.mu.Unlock() + stub.mu.Lock() + stub.dirs["/workspace/"] = true + stub.files["/workspace/probe.txt"] = "x" + stub.mu.Unlock() + + // A running session whose pod has no address yet (what an explicit resume + // leaves behind) must not fail the request outright. + base.mu.Lock() + base.sessions[id].PodIp = "" + base.mu.Unlock() + + listDir(t, ts, s, id) + + if len(base.resumed) != 0 { + t.Fatalf("a running session must not be resumed again, got %d resumes", len(base.resumed)) + } + if base.sessions[id].GetPodIp() == "" { + t.Fatal("the call must have waited for the pod IP to be assigned") + } +} diff --git a/pkg/sandboxmatrix/controller.go b/pkg/sandboxmatrix/controller.go index 656637d52..01b38f15f 100644 --- a/pkg/sandboxmatrix/controller.go +++ b/pkg/sandboxmatrix/controller.go @@ -215,7 +215,7 @@ func reconcileWarmPools(ctx context.Context, k8s kubernetes.Interface, dyn dynam boost = demand.observe(time.Now(), coldStarts) } for _, pool := range pools.Items { - reconcileSinglePool(ctx, k8s, pool, maxPods, cfg, boost) + reconcileSinglePool(ctx, k8s, dyn, pool, maxPods, cfg, boost) } recycleUnhealthyWarmPods(ctx, k8s, cfg.Namespace) updateSandboxMatrixStatus(ctx, k8s, dyn, cfg, orch, maxPods) @@ -357,7 +357,7 @@ func deleteWarmPod(ctx context.Context, k8s kubernetes.Interface, namespace stri } // reconcileSinglePool ensures one WarmPool CRD's target is met within capacity limits. -func reconcileSinglePool(ctx context.Context, k8s kubernetes.Interface, pool unstructured.Unstructured, maxPods int64, cfg config.SandboxConfig, boost int64) { +func reconcileSinglePool(ctx context.Context, k8s kubernetes.Interface, dyn dynamic.Interface, pool unstructured.Unstructured, maxPods int64, cfg config.SandboxConfig, boost int64) { specMap, _ := pool.Object["spec"].(map[string]interface{}) configuredSize, _ := specMap["size"].(int64) runtimeClass, _ := specMap["runtimeClass"].(string) @@ -404,8 +404,10 @@ func reconcileSinglePool(ctx context.Context, k8s kubernetes.Interface, pool uns // returning from resetting, or a target that was just lowered. Scale // down now instead of waiting for the idle reaper's sessionTTL×2 sweep. trimWarmPods(ctx, k8s, cfg.Namespace, warmPods.Items, -gap) + updateWarmPoolStatus(ctx, dyn, cfg, pool, warmPods.Items, 0) return case gap == 0: + updateWarmPoolStatus(ctx, dyn, cfg, pool, warmPods.Items, 0) return } if maxPods >= 0 && int64(len(allPods.Items)) >= maxPods { @@ -419,6 +421,36 @@ func reconcileSinglePool(ctx context.Context, k8s kubernetes.Interface, pool uns logrus.Debugf("sandbox-matrix: create warm pod: %v", err) } } + // The pods we just asked for are still Pending, so report them as pending + // rather than ready: an operator watching the pool sees "1 pending" until + // the next pass flips it to ready, instead of an empty status that looks + // like a broken pool. + updateWarmPoolStatus(ctx, dyn, cfg, pool, warmPods.Items, gap) +} + +// updateWarmPoolStatus writes the pool's ready/pending counts to its .status. +// +// The CRD has carried readyCount/pendingCount since KIP-30 but nothing ever +// populated them, so `kubectl get sandboxwarmpools` reported an empty READY +// column and operators had no in-cluster signal for "is my pool actually +// serving?". Pending is derived from the gap this pass is closing rather than +// counted from pod phases, because a pod created a moment ago has no readable +// phase yet. +func updateWarmPoolStatus(ctx context.Context, dyn dynamic.Interface, cfg config.SandboxConfig, pool unstructured.Unstructured, warmPods []corev1.Pod, pending int64) { + var ready int64 + for i := range warmPods { + if sandboxgrpc.PodReadyCondition(&warmPods[i]) { + ready++ + } + } + status, ok := pool.Object["status"].(map[string]interface{}) + if !ok { + status = map[string]interface{}{} + pool.Object["status"] = status + } + status["readyCount"] = ready + status["pendingCount"] = pending + dyn.Resource(warmPoolGVR).Namespace(cfg.Namespace).UpdateStatus(ctx, &pool, metav1.UpdateOptions{}) //nolint:errcheck } // createWarmPodWithWorkspace provisions a blank workspace PVC and boots a warm diff --git a/pkg/sandboxmatrix/controller_test.go b/pkg/sandboxmatrix/controller_test.go index 4d1557b64..6e17d2b77 100644 --- a/pkg/sandboxmatrix/controller_test.go +++ b/pkg/sandboxmatrix/controller_test.go @@ -364,12 +364,12 @@ func newWarmPoolCR(size int64) unstructured.Unstructured { // lowered target). Surplus must go now, not at the next idle reaper sweep. func TestReconcileSinglePool_TrimsSurplusWarmPods(t *testing.T) { ctx := context.Background() - k8s := kubefake.NewSimpleClientset() + dyn, k8s := fakeSandboxClients() ready := warmTestPod("warm-ready", corev1.PodRunning, condReady) stuck := warmTestPod("warm-stuck", corev1.PodPending, condUnschedulable) seedPods(ctx, t, k8s, ready, stuck) - reconcileSinglePool(ctx, k8s, newWarmPoolCR(1), podCapacityUnknown, defaultCfg(), 0) + reconcileSinglePool(ctx, k8s, dyn, newWarmPoolCR(1), podCapacityUnknown, defaultCfg(), 0) // The unschedulable pod is worthless; the claimable one must stay. assertGone(t, k8s, "warm-stuck", "worthless to a session") @@ -378,13 +378,13 @@ func TestReconcileSinglePool_TrimsSurplusWarmPods(t *testing.T) { func TestReconcileSinglePool_TrimPrefersLongestIdle(t *testing.T) { ctx := context.Background() - k8s := kubefake.NewSimpleClientset() + dyn, k8s := fakeSandboxClients() older := idleSince(warmTestPod("warm-old", corev1.PodRunning, condReady), 90*time.Minute) newer := idleSince(warmTestPod("warm-new", corev1.PodRunning, condReady), 5*time.Minute) seedPods(ctx, t, k8s, older, newer) - reconcileSinglePool(ctx, k8s, newWarmPoolCR(1), podCapacityUnknown, defaultCfg(), 0) + reconcileSinglePool(ctx, k8s, dyn, newWarmPoolCR(1), podCapacityUnknown, defaultCfg(), 0) assertGone(t, k8s, "warm-old", "idled longest") assertPresent(t, k8s, "warm-new") @@ -395,11 +395,11 @@ func TestReconcileSinglePool_TrimPrefersLongestIdle(t *testing.T) { // create a refill the scheduler can never place. func TestReconcileSinglePool_RespectsCapacity(t *testing.T) { ctx := context.Background() - k8s := kubefake.NewSimpleClientset() + dyn, k8s := fakeSandboxClients() active := sandboxTestPod("sandbox-warm-active", sandboxgrpc.StateActive, corev1.PodRunning) seedPods(ctx, t, k8s, active) - reconcileSinglePool(ctx, k8s, newWarmPoolCR(2), 1, defaultCfg(), 0) + reconcileSinglePool(ctx, k8s, dyn, newWarmPoolCR(2), 1, defaultCfg(), 0) pods, err := k8s.CoreV1().Pods(testNS).List(ctx, metav1.ListOptions{}) if err != nil { @@ -746,3 +746,64 @@ func TestGCExpiredSessions_ReclaimsPausedSessionPVC(t *testing.T) { t.Fatalf("expected the paused session's claim %s to be reclaimed, not leaked", claim) } } + +// warmPoolStatusAfterReconcile runs one reconcile pass over a created WarmPool +// CR seeded with the given pods, and returns the .status the reconciler wrote. +// +// Both WarmPool status tests need the same arrange/act/observe shape: create the +// pool CR the reconciler reads, seed the pods it counts, reconcile, then read +// the status back. Repeating that boilerplate in each test is what pushes the +// file's duplication metric over the new-code threshold, and it buries the one +// line each test actually cares about. +func warmPoolStatusAfterReconcile(t *testing.T, poolSize int64, pods ...*corev1.Pod) map[string]interface{} { + t.Helper() + ctx := context.Background() + dyn, k8s := fakeSandboxClients() + seedPods(ctx, t, k8s, pods...) + + pool := newWarmPoolCR(poolSize) + if _, err := dyn.Resource(warmPoolGVR).Namespace(testNS).Create(ctx, &pool, metav1.CreateOptions{}); err != nil { + t.Fatalf("create warm pool CR: %v", err) + } + reconcileSinglePool(ctx, k8s, dyn, pool, podCapacityUnknown, defaultCfg(), 0) + + got, err := dyn.Resource(warmPoolGVR).Namespace(testNS).Get(ctx, "default", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pool: %v", err) + } + status, _ := got.Object["status"].(map[string]interface{}) + if status == nil { + t.Fatal("reconcile must populate .status") + } + return status +} + +// TestReconcileSinglePool_PopulatesStatus is the regression test for the +// WarmPool status surface: the CRD has carried readyCount/pendingCount since +// KIP-30 but nothing ever wrote them, so `kubectl get sandboxwarmpools` showed +// an empty READY column and operators had no in-cluster signal for whether the +// pool was actually serving. +func TestReconcileSinglePool_PopulatesStatus(t *testing.T) { + // One ready warm pod plus the pool's target of 1: the gap is zero, so the + // reconciler should report it as ready rather than pending. + status := warmPoolStatusAfterReconcile(t, 1, + warmTestPod("warm-ready", corev1.PodRunning, condReady)) + + if rc, _ := status["readyCount"].(int64); rc != 1 { + t.Fatalf("readyCount = %v, want 1", status["readyCount"]) + } + if pc, _ := status["pendingCount"].(int64); pc != 0 { + t.Fatalf("pendingCount = %v, want 0", status["pendingCount"]) + } +} + +// TestReconcileSinglePool_ReportsPendingWhenFilling covers the case operators +// hit most: the pool is short and has just created pods that are still Pending, +// so the status must say pending rather than looking empty. +func TestReconcileSinglePool_ReportsPendingWhenFilling(t *testing.T) { + status := warmPoolStatusAfterReconcile(t, 3) + + if _, ok := status["pendingCount"]; !ok { + t.Fatal("pendingCount must be reported while the pool is filling") + } +} diff --git a/sandboxd/src/background.zig b/sandboxd/src/background.zig index 39f5c8507..e6a098108 100644 --- a/sandboxd/src/background.zig +++ b/sandboxd/src/background.zig @@ -118,6 +118,17 @@ pub fn handleBgSubmit(allocator: std.mem.Allocator, client_fd: i32, body: []cons } } + // Parent: reap the wrapper once it exits so it does not linger as a zombie. + // main.setupSignals no longer auto-reaps (a real SIGCHLD handler is what + // keeps exec exit codes observable), so this site now drains its child + // itself. The wait status is deliberately discarded: completion is + // communicated through the exit_code file the BG_WRAPPER EXIT trap writes, + // which handleBgPoll reads. Only the zombie matters here. + { + const reaper = std.Thread.spawn(.{}, reapBackground, .{@as(i32, @intCast(pid))}) catch null; + if (reaper) |t| t.detach(); + } + var resp_buf: [256]u8 = undefined; const resp = try std.fmt.bufPrint(&resp_buf, "{{\"status\":\"started\",\"run_id\":\"{s}\"}}", .{req.run_id}); try main.writeResponse(client_fd, "200 OK", "application/json", resp); @@ -255,6 +266,20 @@ fn unixSeconds() i64 { // run's entire process group (the child called setpgid, so pgid == pid), so a // service started by the wrapper dies with it instead of surviving the kill as // an orphan holding its port. +/// reapBackground blocks until the background wrapper exits and collects +/// its status. Runs detached; exists purely to keep the pid from becoming a +/// zombie now that SIGCHLD no longer auto-reaps. +fn reapBackground(pid: i32) void { + var status: u32 = 0; + _ = std.os.linux.syscall4( + .wait4, + @as(usize, @bitCast(@as(isize, @intCast(pid)))), + @intFromPtr(&status), + 0, + 0, + ); +} + fn spawnTimeoutKiller(pid: i32, timeout_sec: u32, run_dir: [:0]u8, allocator: std.mem.Allocator) void { defer allocator.free(run_dir); const ts = std.os.linux.timespec{ .sec = @as(isize, @intCast(timeout_sec)), .nsec = 0 }; diff --git a/sandboxd/src/exec.zig b/sandboxd/src/exec.zig index 48fa25791..e404f3a70 100644 --- a/sandboxd/src/exec.zig +++ b/sandboxd/src/exec.zig @@ -219,10 +219,15 @@ pub fn runCommandWithEnv(allocator: std.mem.Allocator, command: []const u8, work 0, ); const err = std.posix.errno(@as(usize, @bitCast(rc))); - if (err == .SUCCESS and std.posix.W.IFEXITED(status)) { - break :blk @intCast(std.posix.W.EXITSTATUS(status)); + // A real SIGCHLD handler (main.setupSignals) leaves the status waitable, + // so this recovers it. The old fallback of 0 reported success for a + // killed command, which made `cmd; echo $?` lie to every caller. + if (err != .SUCCESS) break :blk -1; + if (std.posix.W.IFEXITED(status)) break :blk @intCast(std.posix.W.EXITSTATUS(status)); + if (std.posix.W.IFSIGNALED(status)) { + break :blk 128 + @as(i32, @intCast(@as(u8, @truncate(status)))); } - break :blk 0; + break :blk -1; }; var ts_end: std.os.linux.timespec = undefined; @@ -416,9 +421,12 @@ pub fn handleExec(allocator: std.mem.Allocator, client_fd: i32, body: []const u8 _ = std.os.linux.close(stdout_pipe[0]); execctl.markDone(@intCast(child_pid)); - // Reap child, then drop the process-table entry (closes stdin fd). - // The exit code is reported as the final SSE frame so the e2b layer - // gets the honest code in-stream (retiring the marker-file hack). + // Reap the child and report its exit code as the final SSE frame so + // the e2b layer gets the honest code in-stream. + // + // wait4 succeeds here because setupSignals installs a real SIGCHLD + // handler (not SIG_IGN/SA_NOCLDWAIT), so the kernel leaves the status + // word for us instead of auto-reaping it. var status: u32 = 0; const wrc = std.os.linux.syscall4( .wait4, @@ -428,10 +436,17 @@ pub fn handleExec(allocator: std.mem.Allocator, client_fd: i32, body: []const u8 0, ); const werr = std.posix.errno(@as(usize, @bitCast(wrc))); - const exit_code: i32 = if (werr == .SUCCESS and std.posix.W.IFEXITED(status)) + const exit_code: i32 = if (werr != .SUCCESS) + // Genuinely unreapable (should not happen with a real handler). + -1 + else if (std.posix.W.IFEXITED(status)) @intCast(std.posix.W.EXITSTATUS(status)) + else if (std.posix.W.IFSIGNALED(status)) + // Report the terminating signal the way a shell does: 128+signo, + // so `kill -9` surfaces as 137 rather than a bare -1. + 128 + @as(i32, @intCast(@as(u8, @truncate(status)))) else - -1; // killed by signal / not reaped normally + -1; var exit_frame_buf: [64]u8 = undefined; const exit_frame = std.fmt.bufPrint(&exit_frame_buf, "data: {{\"exit\":{d}}}\n\n", .{exit_code}) catch { _ = std.os.linux.write(client_fd, "data: {\"exit\":-1}\n\n".ptr, 22); @@ -463,9 +478,7 @@ pub fn handleExec(allocator: std.mem.Allocator, client_fd: i32, body: []const u8 defer allocator.free(stderr_json); const trunc_lit: []const u8 = if (result.truncated) "true" else "false"; - const resp = try std.fmt.allocPrint(allocator, - "{{\"stdout\":\"{s}\",\"stderr\":\"{s}\",\"exit_code\":{d},\"duration_ms\":{d},\"truncated\":{s}}}", - .{ stdout_json, stderr_json, result.exit_code, result.duration_ms, trunc_lit }); + const resp = try std.fmt.allocPrint(allocator, "{{\"stdout\":\"{s}\",\"stderr\":\"{s}\",\"exit_code\":{d},\"duration_ms\":{d},\"truncated\":{s}}}", .{ stdout_json, stderr_json, result.exit_code, result.duration_ms, trunc_lit }); defer allocator.free(resp); try main.writeResponse(client_fd, "200 OK", "application/json", resp); } diff --git a/sandboxd/src/main.zig b/sandboxd/src/main.zig index 11b1e6080..f5f681c9c 100644 --- a/sandboxd/src/main.zig +++ b/sandboxd/src/main.zig @@ -23,7 +23,9 @@ pub fn main() !void { gpa_state = .{}; break :blk gpa_state.allocator(); } else std.heap.page_allocator; - defer { if (std.debug.runtime_safety) _ = gpa_state.deinit(); } + defer { + if (std.debug.runtime_safety) _ = gpa_state.deinit(); + } // PID 1: reap zombies via SIGCHLD ignore const pid = std.os.linux.getpid(); @@ -197,8 +199,7 @@ fn handleRequest(allocator: std.mem.Allocator, client_fd: i32) !void { /// require status=="ready". fn handleReady(client_fd: i32) !void { const venv_ready = venv.isReady(); - const body = try std.fmt.allocPrint(std.heap.page_allocator, - "{{\"status\":\"ready\",\"venv\":{s}}}", .{if (venv_ready) "true" else "false"}); + const body = try std.fmt.allocPrint(std.heap.page_allocator, "{{\"status\":\"ready\",\"venv\":{s}}}", .{if (venv_ready) "true" else "false"}); defer std.heap.page_allocator.free(body); try writeResponse(client_fd, "200 OK", "application/json", body); } @@ -246,19 +247,34 @@ fn handleEvents(allocator: std.mem.Allocator, client_fd: i32, query: []const u8) /// sandboxd is linux-only and writes via raw syscalls. pub fn writeResponse(client_fd: i32, status: []const u8, content_type: []const u8, body: []const u8) !void { var header_buf: [4096]u8 = undefined; - const header = try std.fmt.bufPrint(&header_buf, - "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n", - .{ status, content_type, body.len }); + const header = try std.fmt.bufPrint(&header_buf, "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n", .{ status, content_type, body.len }); _ = std.os.linux.write(client_fd, header.ptr, header.len); _ = std.os.linux.write(client_fd, body.ptr, body.len); } fn setupSignals() void { - // Ignore SIGCHLD to auto-reap zombies as PID 1 + // Install a REAL (non-SIG_IGN) SIGCHLD handler rather than SIG_IGN + + // SA_NOCLDWAIT. Both SIG_IGN and SA_NOCLDWAIT make the kernel auto-reap + // children, which means wait4 on any child returns -ECHILD and leaves the + // status word untouched — every exec's exit code silently collapsed to the + // caller's fallback (-1 on the streaming path, 0 on the buffered one). + // + // A real handler keeps children waitable, so the exit-code paths in + // exec.zig / pty.zig / venv.zig recover the real status. Zombie reaping is + // then each fork site's responsibility: exec, pty and venv already wait, + // and background.zig now drains its own child (see reapBackground). + // + // SA_RESTART keeps the daemon's blocking syscalls (accept, read) from + // failing with EINTR every time a command finishes. const sa = std.os.linux.Sigaction{ - .handler = .{ .handler = std.os.linux.SIG.IGN }, + .handler = .{ .handler = sigchldNoop }, .mask = std.mem.zeroes(std.os.linux.sigset_t), - .flags = std.os.linux.SA.NOCLDWAIT, + .flags = std.os.linux.SA.RESTART, }; _ = std.os.linux.sigaction(std.os.linux.SIG.CHLD, &sa, null); } + +/// sigchldNoop receives SIGCHLD so the kernel leaves the child's wait status +/// for its parent to collect. It does no work: reaping happens at each fork +/// site's explicit wait, which is what lets the exit code survive. +fn sigchldNoop(_: std.os.linux.SIG) callconv(.c) void {} diff --git a/sandboxd/src/pty.zig b/sandboxd/src/pty.zig index 1c4f8592c..9629007de 100644 --- a/sandboxd/src/pty.zig +++ b/sandboxd/src/pty.zig @@ -285,9 +285,15 @@ fn reapExitCode(pid: i32) i32 { 0, 0, ); - if (isErr(rc)) return 0; // auto-reaped (SA_NOCLDWAIT) → no exit status + // A real SIGCHLD handler keeps the status waitable (see main.setupSignals), + // so the old "auto-reaped → return 0" branch no longer fires and PTY + // sessions report their true exit code. + if (isErr(rc)) return -1; if (std.posix.W.IFEXITED(status)) return @intCast(std.posix.W.EXITSTATUS(status)); - return 0; + if (std.posix.W.IFSIGNALED(status)) { + return 128 + @as(i32, @intCast(@as(u8, @truncate(status)))); + } + return -1; } fn pumpThread(id: u32, master_fd: i32, pid: i32) void { diff --git a/tests/e2b_official_aligned.py b/tests/e2b_official_aligned.py new file mode 100644 index 000000000..64a6b9c37 --- /dev/null +++ b/tests/e2b_official_aligned.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""E2B compatibility check aligned with https://docs.e2b.dev/client and the +official lifecycle / persistence / quickstart examples. + +Every step below is a verbatim-shape official example (create, commands.run, +files.write/read/list, set_timeout, get_info, Sandbox.list, pause, kill), plus +`connect()` — the documented replacement for the non-existent `resume()`. + +Usage: + E2B_API_URL=http://:3676 \ + E2B_API_KEY=e2b_ python3 e2b_official_aligned.py +""" +import os +import sys +import traceback + +from e2b import Sandbox + +API_URL = os.environ.get("E2B_API_URL", "http://43.203.225.109:3676") + +results = [] + + +def check(name, fn): + """Run one official-example step and record PASS/FAIL + detail.""" + try: + detail = fn() + results.append((name, True, detail)) + print(f"[PASS] {name}: {detail}") + except Exception as exc: # noqa: BLE001 - report, do not abort the suite + tb = traceback.format_exc() + results.append((name, False, f"{type(exc).__name__}: {exc}")) + print(f"[FAIL] {name}: {type(exc).__name__}: {exc}") + print(tb) + + +# ── L1: Sandbox.create ─────────────────────────────────────────────────────── +sandbox = None + + +def step_create(): + global sandbox + sandbox = Sandbox.create(timeout=60) + return f"sandbox_id={sandbox.sandbox_id}" + + +# ── L1: commands.run (official quickstart shape) ───────────────────────────── +def step_run(): + r = sandbox.commands.run('python3 -c "print(40+2)"') + return (f"stdout={r.stdout!r} stderr={r.stderr!r} " + f"exit_code={r.exit_code} error={r.error!r}") + + +def step_run_exit_code(): + """stdout AND exit code must both be right — the k8e2 regression.""" + r = sandbox.commands.run("exit 7") + return f"stdout={r.stdout!r} exit_code={r.exit_code} (expected 7)" + + +def step_run_stderr(): + r = sandbox.commands.run("echo out; echo err >&2; exit 3") + return (f"stdout={r.stdout!r} stderr={r.stderr!r} " + f"exit_code={r.exit_code} (expected 3)") + + +# ── L2: files.write / files.read (official quickstart shape) ──────────────── +def step_files_write_read(): + sandbox.files.write("/tmp/e2b-check.txt", "hello from official example\n") + got = sandbox.files.read("/tmp/e2b-check.txt") + return f"read back {got!r}" + + +def step_files_list(): + entries = sandbox.files.list("/tmp") + names = [getattr(e, "name", None) or getattr(e, "path", None) for e in entries] + hit = [n for n in names if n and "e2b-check" in n] + return f"{len(names)} entries, found {hit}" + + +# ── L3: set_timeout ───────────────────────────────────────────────────────── +def step_set_timeout(): + sandbox.set_timeout(120) + return "set_timeout(120) ok" + + +def step_set_timeout_never(): + sandbox.set_timeout(-1) + return "set_timeout(-1) ok" + + +# ── L3: get_info / Sandbox.list ───────────────────────────────────────────── +def step_get_info(): + info = sandbox.get_info() + return (f"state={getattr(info, 'state', None)!r} " + f"end_at={getattr(info, 'end_at', None)!r} " + f"sandbox_id={getattr(info, 'sandbox_id', None)!r}") + + +def step_list(): + items = Sandbox.list().next_items() + mine = [s for s in items if s.sandbox_id == sandbox.sandbox_id] + return f"{len(items)} total, ours present={bool(mine)}" + + +# ── L3: pause (persistence doc shape) ─────────────────────────────────────── +def step_pause(): + sandbox.pause() + return "paused" + + +# ── L3: connect — the DOCUMENTED resume path (no resume() in SDK 2.39.1) ──── +def step_connect(): + same = sandbox.connect() + return f"reconnected sandbox_id={same.sandbox_id}" + + +def step_connect_files_survive(): + """A real persistence check: data written before pause survives connect.""" + same = sandbox.connect() + got = same.files.read("/tmp/e2b-check.txt") + return f"state survived reconnect: {got!r}" + + +# ── L4: watch_dir (envd filesystem watcher) ───────────────────────────────── +def step_watch_dir(): + h = sandbox.files.watch_dir("/tmp") + # write inside the watched dir, then poll for the event + sandbox.files.write("/tmp/e2b-watched.txt", "x") + import time + deadline = time.time() + 15 + seen = [] + while time.time() < deadline: + for ev in h.get_new_events(): + seen.append(getattr(ev, "name", None) or str(ev)) + if seen: + break + time.sleep(0.5) + h.stop() + return f"events={seen}" if seen else "no events within 15s" + + +# ── Cleanup: kill ─────────────────────────────────────────────────────────── +def step_kill(): + sandbox.kill() + return "killed" + + +ORDER = [ + ("L1 create", step_create), + ("L1 commands.run stdout+exit_code", step_run), + ("L1 commands.run pure exit code", step_run_exit_code), + ("L1 commands.run stderr+exit code", step_run_stderr), + ("L2 files.write / files.read", step_files_write_read), + ("L2 files.list", step_files_list), + ("L3 set_timeout(120)", step_set_timeout), + ("L3 set_timeout(-1)", step_set_timeout_never), + ("L3 get_info", step_get_info), + ("L3 Sandbox.list", step_list), + ("L3 pause", step_pause), + ("L3 connect (documented resume path)", step_connect), + ("L3 connect preserves files", step_connect_files_survive), + ("L4 watch_dir", step_watch_dir), + ("Cleanup kill", step_kill), +] + + +def main(): + print(f"# E2B official-example-aligned compatibility check") + print(f"# endpoint: {API_URL}\n") + for name, fn in ORDER: + if sandbox is None and name != "L1 create": + print(f"[SKIP] {name}: no sandbox") + results.append((name, None, "skipped - no sandbox")) + continue + check(name, fn) + + passed = sum(1 for _, ok, _ in results if ok is True) + failed = sum(1 for _, ok, _ in results if ok is False) + skipped = sum(1 for _, ok, _ in results if ok is None) + print(f"\n# summary: {passed} pass, {failed} fail, {skipped} skip") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main())