diff --git a/.gitignore b/.gitignore index 39a9d5e..957d632 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,6 @@ test/e2e/python/.pytest_cache/ # TypeScript E2E test artifacts test/e2e/ts/node_modules/ test/e2e/ts/package-lock.json + +# Vendored dependencies +vendor/ diff --git a/cmd/edvabe/main.go b/cmd/edvabe/main.go index 6298940..73af049 100644 --- a/cmd/edvabe/main.go +++ b/cmd/edvabe/main.go @@ -164,13 +164,15 @@ func serveCmd(args []string) { domain = fmt.Sprintf("localhost:%d", *port) } mgr, err := sandbox.NewManager(sandbox.Options{ - Runtime: rt, - Agent: ap, - Domain: domain, - Resolver: template.NewSandboxResolver(templateStore), - FreezeDuration: *freezeDuration, - MaxFrozen: *maxFrozen, - StoppedGCAfter: *stoppedGCAfter, + Runtime: rt, + Agent: ap, + Domain: domain, + Resolver: template.NewSandboxResolver(templateStore), + FreezeDuration: *freezeDuration, + MaxFrozen: *maxFrozen, + StoppedGCAfter: *stoppedGCAfter, + KeepaliveEnabled: os.Getenv("EDVABE_KEEPALIVE_ENABLED") != "0", + KeepaliveCoalesce: envDurationOr("EDVABE_KEEPALIVE_COALESCE", time.Second), }) if err != nil { fmt.Fprintf(os.Stderr, "serve: init manager: %v\n", err) @@ -224,13 +226,13 @@ func serveCmd(args []string) { FileSigner: fileSigner, PublicBase: os.Getenv("EDVABE_PUBLIC_BASE"), }) - proxyHandler := api.NewProxy(mgr, rt) + proxyHandler := api.NewProxy(mgr, rt, mgr) dashboardHandler := dashboard.NewHandler(dashboard.HandlerOptions{ Manager: mgr, Runtime: rt, Templates: templateStore, }) - handler := api.NewRouter(controlHandler, proxyHandler, dashboardHandler) + handler := api.NewRouter(controlHandler, proxyHandler, dashboardHandler, mgr.MarkActivity) if *dnsListen != "" { answerStr := *dnsAnswer diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 4ee4f2e..d553993 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -71,10 +71,14 @@ type InitConfig struct { HyperloopIP string } -// VolumeMount describes a filesystem path inside the sandbox backed by -// an external volume. Phase 1 passes an empty list; Phase 4 will wire -// real volumes through the control plane. +// VolumeMount describes an NFS-backed filesystem path inside the +// sandbox, as expected by upstream envd's /init endpoint. edvabe +// does NOT use this — local Docker volumes are mounted at container +// creation time by the runtime, not by envd. This struct exists +// solely for wire-parity with envd's init body and must never be +// populated with local Docker volume names. See +// docs/08-volume-mounts-plan.md § Envd boundary. type VolumeMount struct { - Name string - MountPath string + NFSTarget string `json:"nfs_target,omitempty"` + Path string `json:"path,omitempty"` } diff --git a/internal/agent/upstream/provider_test.go b/internal/agent/upstream/provider_test.go index 5698006..50664e6 100644 --- a/internal/agent/upstream/provider_test.go +++ b/internal/agent/upstream/provider_test.go @@ -70,7 +70,7 @@ func TestInitAgentPostsExpectedPayload(t *testing.T) { EnvVars: map[string]string{"FOO": "bar"}, DefaultUser: "user", DefaultWorkdir: "/home/user", - VolumeMounts: []agent.VolumeMount{{Name: "vol", MountPath: "/volumes/vol"}}, + VolumeMounts: []agent.VolumeMount{{NFSTarget: "server:/export", Path: "/volumes/vol"}}, }) if err != nil { t.Fatalf("InitAgent: %v", err) @@ -87,7 +87,7 @@ func TestInitAgentPostsExpectedPayload(t *testing.T) { if got.DefaultWorkdir != "/home/user" { t.Errorf("DefaultWorkdir = %q", got.DefaultWorkdir) } - if len(got.VolumeMounts) != 1 || got.VolumeMounts[0].MountPath != "/volumes/vol" { + if len(got.VolumeMounts) != 1 || got.VolumeMounts[0].Path != "/volumes/vol" { t.Errorf("VolumeMounts = %#v", got.VolumeMounts) } if got.Timestamp == "" { diff --git a/internal/api/control/router.go b/internal/api/control/router.go index 40a5f0f..d7becbf 100644 --- a/internal/api/control/router.go +++ b/internal/api/control/router.go @@ -61,12 +61,13 @@ func NewRouter(opts RouterOptions) http.Handler { // lifetime — no persistence across restarts. apiKeys := newAPIKeyStore() accessTokens := newAccessTokenStore() - volumes := newVolumeStore() + volumes := newVolumeStore(opts.Runtime) + _ = volumes.loadFromRuntime(context.Background()) sandboxHandler := api.RequireAPIKey(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodPost && r.URL.Path == "/sandboxes": - createSandbox(opts.Manager, opts.Provider, w, r) + createSandbox(opts.Manager, opts.Provider, volumes, w, r) case r.Method == http.MethodGet && r.URL.Path == "/v2/sandboxes": listSandboxes(opts.Manager, opts.Provider, w, r) diff --git a/internal/api/control/sandboxes.go b/internal/api/control/sandboxes.go index 4c5e1f7..9bcdcdd 100644 --- a/internal/api/control/sandboxes.go +++ b/internal/api/control/sandboxes.go @@ -3,7 +3,9 @@ package control import ( "encoding/json" "errors" + "fmt" "net/http" + "path" "sort" "strconv" "strings" @@ -28,12 +30,17 @@ type newSandboxRequest struct { CPUCount int `json:"cpuCount,omitempty"` MemoryMB int `json:"memoryMB,omitempty"` // Accepted for wire compat but not enforced in edvabe. - Secure bool `json:"secure,omitempty"` - AllowInternetAccess *bool `json:"allow_internet_access,omitempty"` - Network *json.RawMessage `json:"network,omitempty"` - VolumeMounts []json.RawMessage `json:"volumeMounts,omitempty"` - MCP *json.RawMessage `json:"mcp,omitempty"` - AutoResume *json.RawMessage `json:"autoResume,omitempty"` + Secure bool `json:"secure,omitempty"` + AllowInternetAccess *bool `json:"allow_internet_access,omitempty"` + Network *json.RawMessage `json:"network,omitempty"` + VolumeMounts []sandboxVolumeMount `json:"volumeMounts,omitempty"` + MCP *json.RawMessage `json:"mcp,omitempty"` + AutoResume *json.RawMessage `json:"autoResume,omitempty"` +} + +type sandboxVolumeMount struct { + Name string `json:"name"` + Path string `json:"path"` } // newSandboxLifecycleInput mirrors the SDK's NewSandbox.lifecycle field. @@ -56,18 +63,26 @@ type sandboxResponse struct { Metadata map[string]string `json:"metadata,omitempty"` StartedAt time.Time `json:"startedAt"` EndAt time.Time `json:"endAt"` + LastActiveAt time.Time `json:"lastActiveAt,omitempty"` } type sandboxDetailResponse struct { sandboxResponse - State sandbox.State `json:"state"` - CPUCount int `json:"cpuCount"` - MemoryMB int64 `json:"memoryMB"` - DiskSizeMB int64 `json:"diskSizeMB"` - Lifecycle sandboxLifecycle `json:"lifecycle"` - Network sandboxNetworkConfig `json:"network"` - AllowInternetAccess bool `json:"allowInternetAccess"` - EnvVars map[string]string `json:"envVars,omitempty"` + State sandbox.State `json:"state"` + CPUCount int `json:"cpuCount"` + MemoryMB int64 `json:"memoryMB"` + DiskSizeMB int64 `json:"diskSizeMB"` + Lifecycle sandboxLifecycle `json:"lifecycle"` + Network sandboxNetworkConfig `json:"network"` + AllowInternetAccess bool `json:"allowInternetAccess"` + EnvVars map[string]string `json:"envVars,omitempty"` + LastActiveAt time.Time `json:"lastActiveAt"` + VolumeMounts []volumeMountResponse `json:"volumeMounts"` +} + +type volumeMountResponse struct { + Name string `json:"name"` + Path string `json:"path"` } type sandboxLifecycle struct { @@ -89,22 +104,33 @@ type timeoutRequest struct { Timeout int `json:"timeout"` } -func createSandbox(manager sandboxManager, provider versionProvider, w http.ResponseWriter, r *http.Request) { +func createSandbox(manager sandboxManager, provider versionProvider, volumes *volumeStore, w http.ResponseWriter, r *http.Request) { var req newSandboxRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { api.WriteError(w, http.StatusBadRequest, "invalid JSON body") return } + var volumeMounts []sandbox.VolumeMount + if len(req.VolumeMounts) > 0 { + resolved, err := resolveVolumeMounts(req.VolumeMounts, volumes) + if err != nil { + api.WriteError(w, http.StatusBadRequest, err.Error()) + return + } + volumeMounts = resolved + } + sbx, err := manager.Create(r.Context(), sandbox.CreateOptions{ - TemplateID: req.TemplateID, - Alias: req.Alias, - Metadata: req.Metadata, - EnvVars: req.EnvVars, - Timeout: time.Duration(req.Timeout) * time.Second, - OnTimeout: resolveOnTimeout(req), - CPUCount: req.CPUCount, - MemoryMB: req.MemoryMB, + TemplateID: req.TemplateID, + Alias: req.Alias, + Metadata: req.Metadata, + EnvVars: req.EnvVars, + Timeout: time.Duration(req.Timeout) * time.Second, + OnTimeout: resolveOnTimeout(req), + CPUCount: req.CPUCount, + MemoryMB: req.MemoryMB, + VolumeMounts: volumeMounts, }) if err != nil { api.WriteError(w, http.StatusInternalServerError, err.Error()) @@ -439,13 +465,13 @@ func getSandboxLogs(manager sandboxManager, w http.ResponseWriter, r *http.Reque // ────────────────────────────────────────────────────────────────────── type metricsEntry struct { - Timestamp time.Time `json:"timestamp"` - CPUCount int `json:"cpuCount"` - CPUUsedPct float64 `json:"cpuUsedPct"` - MemTotalMiB int64 `json:"memTotalMiB"` - MemUsedMiB int64 `json:"memUsedMiB"` - DiskTotalMiB int64 `json:"diskTotalMiB"` - DiskUsedMiB int64 `json:"diskUsedMiB"` + Timestamp time.Time `json:"timestamp"` + CPUCount int `json:"cpuCount"` + CPUUsedPct float64 `json:"cpuUsedPct"` + MemTotalMiB int64 `json:"memTotalMiB"` + MemUsedMiB int64 `json:"memUsedMiB"` + DiskTotalMiB int64 `json:"diskTotalMiB"` + DiskUsedMiB int64 `json:"diskUsedMiB"` } func getSandboxMetrics(manager sandboxManager, rt runtime.Runtime, w http.ResponseWriter, r *http.Request) { @@ -465,11 +491,11 @@ func getSandboxMetrics(manager sandboxManager, rt runtime.Runtime, w http.Respon return } writeJSON(w, http.StatusOK, []metricsEntry{{ - Timestamp: time.Now(), - CPUUsedPct: stats.CPUUsedPercent, - MemTotalMiB: stats.MemoryLimitMB, - MemUsedMiB: stats.MemoryUsedMB, - DiskUsedMiB: stats.DiskUsedMB, + Timestamp: time.Now(), + CPUUsedPct: stats.CPUUsedPercent, + MemTotalMiB: stats.MemoryLimitMB, + MemUsedMiB: stats.MemoryUsedMB, + DiskUsedMiB: stats.DiskUsedMB, }}) } @@ -549,7 +575,8 @@ func toSandboxResponse(manager sandboxManager, provider versionProvider, sbx *sa Domain: manager.Domain(), Metadata: sbx.Metadata, StartedAt: sbx.CreatedAt, - EndAt: sbx.ExpiresAt, + EndAt: sbx.ExpiresAt(), + LastActiveAt: sbx.LastActiveAt, } } @@ -567,6 +594,8 @@ func toSandboxDetailResponse(manager sandboxManager, provider versionProvider, s Network: sandboxNetworkConfig{AllowPublicTraffic: true, AllowOut: []string{}, DenyOut: []string{}}, AllowInternetAccess: true, EnvVars: sbx.EnvVars, + LastActiveAt: sbx.LastActiveAt, + VolumeMounts: toVolumeMountResponse(sbx.VolumeMounts), } // Fall back to runtime stats for sandboxes created before this // field existed (CPUCount/MemoryMB == 0 means "unlimited" on the @@ -580,6 +609,17 @@ func toSandboxDetailResponse(manager sandboxManager, provider versionProvider, s return resp } +func toVolumeMountResponse(mounts []sandbox.VolumeMount) []volumeMountResponse { + if len(mounts) == 0 { + return []volumeMountResponse{} + } + out := make([]volumeMountResponse, len(mounts)) + for i, m := range mounts { + out[i] = volumeMountResponse{Name: m.Name, Path: m.Path} + } + return out +} + // resolveOnTimeout picks the sandbox lifecycle mode from the incoming // request. Both autoPause (legacy shorthand) and lifecycle.onTimeout // are accepted; lifecycle.onTimeout wins when both are set, mirroring @@ -599,3 +639,45 @@ func resolveOnTimeout(req newSandboxRequest) sandbox.OnTimeoutMode { } return sandbox.OnTimeoutKill } + +const maxVolumeMounts = 16 + +// resolveVolumeMounts validates each volumeMount entry and resolves +// the logical volume name to a Docker volume source. Returns an error +// if any mount is invalid or references an unknown volume. +func resolveVolumeMounts(mounts []sandboxVolumeMount, volumes *volumeStore) ([]sandbox.VolumeMount, error) { + if len(mounts) > maxVolumeMounts { + return nil, fmt.Errorf("too many volume mounts (max %d)", maxVolumeMounts) + } + seen := make(map[string]bool, len(mounts)) + out := make([]sandbox.VolumeMount, 0, len(mounts)) + for _, m := range mounts { + if m.Name == "" { + return nil, errors.New("volumeMount.name is required") + } + if m.Path == "" { + return nil, errors.New("volumeMount.path is required") + } + cleaned := path.Clean(m.Path) + if !path.IsAbs(cleaned) { + return nil, fmt.Errorf("volumeMount.path %q is not absolute", m.Path) + } + if cleaned == "/" { + return nil, errors.New("volumeMount.path cannot be /") + } + if seen[cleaned] { + return nil, fmt.Errorf("duplicate volumeMount.path %q", cleaned) + } + rec, ok := volumes.resolveByName(m.Name) + if !ok { + return nil, fmt.Errorf("volume %q not found", m.Name) + } + seen[cleaned] = true + out = append(out, sandbox.VolumeMount{ + Name: m.Name, + Path: cleaned, + DockerName: dockerNameForVolumeID(rec.VolumeID), + }) + } + return out, nil +} diff --git a/internal/api/control/stubs.go b/internal/api/control/stubs.go index fc9ef8a..1ca10c7 100644 --- a/internal/api/control/stubs.go +++ b/internal/api/control/stubs.go @@ -165,73 +165,9 @@ func (s *accessTokenStore) delete(w http.ResponseWriter, r *http.Request) { } // ────────────────────────────────────────────────────────────────────── -// Volumes — in-memory registry. No real bind-mount in v1. +// Volumes — moved to volumes.go (runtime-backed). // ────────────────────────────────────────────────────────────────────── -type volumeEntry struct { - VolumeID string `json:"volumeID"` - Name string `json:"name"` -} - -type volumeStore struct { - volumes []volumeEntry -} - -func newVolumeStore() *volumeStore { - return &volumeStore{} -} - -func (s *volumeStore) list(w http.ResponseWriter, _ *http.Request) { - out := s.volumes - if out == nil { - out = []volumeEntry{} - } - writeJSON(w, http.StatusOK, out) -} - -func (s *volumeStore) create(w http.ResponseWriter, r *http.Request) { - var req struct { - Name string `json:"name"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - api.WriteError(w, http.StatusBadRequest, "invalid JSON body") - return - } - entry := volumeEntry{ - VolumeID: "vol_" + randomHex(8), - Name: req.Name, - } - s.volumes = append(s.volumes, entry) - writeJSON(w, http.StatusCreated, entry) -} - -func (s *volumeStore) get(w http.ResponseWriter, r *http.Request) { - id := strings.TrimPrefix(r.URL.Path, "/volumes/") - for _, v := range s.volumes { - if v.VolumeID == id || v.Name == id { - writeJSON(w, http.StatusOK, map[string]any{ - "volumeID": v.VolumeID, - "name": v.Name, - "token": "fake-volume-jwt", - }) - return - } - } - api.WriteError(w, http.StatusNotFound, "volume not found") -} - -func (s *volumeStore) delete(w http.ResponseWriter, r *http.Request) { - id := strings.TrimPrefix(r.URL.Path, "/volumes/") - for i, v := range s.volumes { - if v.VolumeID == id || v.Name == id { - s.volumes = append(s.volumes[:i], s.volumes[i+1:]...) - w.WriteHeader(http.StatusNoContent) - return - } - } - api.WriteError(w, http.StatusNotFound, "volume not found") -} - // ────────────────────────────────────────────────────────────────────── // Snapshots list — returns empty array. Individual snapshot creation // lives in sandboxes.go (POST /sandboxes/{id}/snapshots). diff --git a/internal/api/control/stubs_test.go b/internal/api/control/stubs_test.go index b712ebb..43de3a1 100644 --- a/internal/api/control/stubs_test.go +++ b/internal/api/control/stubs_test.go @@ -3,6 +3,7 @@ package control import ( "bytes" "encoding/json" + "io" "net/http" "net/http/httptest" "strings" @@ -131,7 +132,7 @@ func TestVolumesLifecycle(t *testing.T) { // List empty resp := apiGet(t, srv.URL+"/volumes") - var vols []volumeEntry + var vols []volumeResponse json.NewDecoder(resp.Body).Decode(&vols) resp.Body.Close() if len(vols) != 0 { @@ -144,7 +145,7 @@ func TestVolumesLifecycle(t *testing.T) { if resp2.StatusCode != 201 { t.Fatalf("create status = %d", resp2.StatusCode) } - var created volumeEntry + var created volumeResponse json.NewDecoder(resp2.Body).Decode(&created) resp2.Body.Close() if created.Name != "data-vol" { @@ -323,6 +324,120 @@ func TestCreateSandboxWithExtraFieldsDoesNotCrash(t *testing.T) { } } +func TestCreateSandboxWithVolumeMounts(t *testing.T) { + srv := httptest.NewServer(newTestControlRouter(t)) + defer srv.Close() + + // Create a volume first. + volBody, _ := json.Marshal(map[string]string{"name": "test-vol"}) + resp := apiPost(t, srv.URL+"/volumes", volBody) + if resp.StatusCode != 201 { + t.Fatalf("volume create status = %d", resp.StatusCode) + } + var vol volumeResponse + json.NewDecoder(resp.Body).Decode(&vol) + resp.Body.Close() + if vol.Name != "test-vol" { + t.Fatalf("volume name = %q, want test-vol", vol.Name) + } + + // Create a sandbox with the volume mounted. + sbxBody, _ := json.Marshal(map[string]any{ + "templateID": "base", + "timeout": 60, + "volumeMounts": []map[string]string{ + {"name": "test-vol", "path": "/mnt/data"}, + }, + }) + resp2 := apiPost(t, srv.URL+"/sandboxes", sbxBody) + if resp2.StatusCode != 201 { + t.Fatalf("sandbox create status = %d, body=%s", resp2.StatusCode, readBody(resp2)) + } + var created sandboxResponse + json.NewDecoder(resp2.Body).Decode(&created) + resp2.Body.Close() + + // GET the sandbox detail and verify volumeMounts is echoed. + resp3 := apiGet(t, srv.URL+"/sandboxes/"+created.SandboxID) + if resp3.StatusCode != 200 { + t.Fatalf("get status = %d", resp3.StatusCode) + } + var detail sandboxDetailResponse + json.NewDecoder(resp3.Body).Decode(&detail) + resp3.Body.Close() + if len(detail.VolumeMounts) != 1 { + t.Fatalf("expected 1 volume mount, got %d", len(detail.VolumeMounts)) + } + if detail.VolumeMounts[0].Name != "test-vol" || detail.VolumeMounts[0].Path != "/mnt/data" { + t.Errorf("volumeMount = %+v, want {test-vol /mnt/data}", detail.VolumeMounts[0]) + } +} + +func TestCreateSandboxRejectsUnknownVolume(t *testing.T) { + srv := httptest.NewServer(newTestControlRouter(t)) + defer srv.Close() + + sbxBody, _ := json.Marshal(map[string]any{ + "templateID": "base", + "timeout": 60, + "volumeMounts": []map[string]string{ + {"name": "nonexistent", "path": "/mnt/data"}, + }, + }) + resp := apiPost(t, srv.URL+"/sandboxes", sbxBody) + if resp.StatusCode != 400 { + t.Fatalf("expected 400 for unknown volume, got %d", resp.StatusCode) + } + resp.Body.Close() +} + +func TestCreateSandboxRejectsInvalidVolumeMounts(t *testing.T) { + srv := httptest.NewServer(newTestControlRouter(t)) + defer srv.Close() + + // Create a valid volume first. + volBody, _ := json.Marshal(map[string]string{"name": "valid-vol"}) + resp := apiPost(t, srv.URL+"/volumes", volBody) + resp.Body.Close() + + tests := []struct { + name string + body string + }{ + {"empty name", `{"templateID":"base","volumeMounts":[{"name":"","path":"/mnt/data"}]}`}, + {"relative path", `{"templateID":"base","volumeMounts":[{"name":"valid-vol","path":"relative/path"}]}`}, + {"root path", `{"templateID":"base","volumeMounts":[{"name":"valid-vol","path":"/"}]}`}, + {"duplicate path", `{"templateID":"base","volumeMounts":[{"name":"valid-vol","path":"/mnt/a"},{"name":"valid-vol","path":"/mnt/a"}]}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resp := apiPost(t, srv.URL+"/sandboxes", []byte(tt.body)) + if resp.StatusCode != 400 { + t.Errorf("expected 400 for %s, got %d", tt.name, resp.StatusCode) + } + resp.Body.Close() + }) + } +} + +func TestVolumeDuplicateName(t *testing.T) { + srv := httptest.NewServer(newTestControlRouter(t)) + defer srv.Close() + + body1, _ := json.Marshal(map[string]string{"name": "dupe-vol"}) + resp1 := apiPost(t, srv.URL+"/volumes", body1) + if resp1.StatusCode != 201 { + t.Fatalf("first create status = %d", resp1.StatusCode) + } + resp1.Body.Close() + + resp2 := apiPost(t, srv.URL+"/volumes", body1) + if resp2.StatusCode != 409 { + t.Fatalf("duplicate create status = %d, want 409", resp2.StatusCode) + } + resp2.Body.Close() +} + // ── Helpers ────────────────────────────────────────────────────────── func apiGet(t *testing.T, url string) *http.Response { @@ -353,3 +468,8 @@ func apiPost(t *testing.T, url string, body []byte) *http.Response { } return resp } + +func readBody(resp *http.Response) string { + b, _ := io.ReadAll(resp.Body) + return string(b) +} diff --git a/internal/api/control/volumes.go b/internal/api/control/volumes.go new file mode 100644 index 0000000..4cfdb34 --- /dev/null +++ b/internal/api/control/volumes.go @@ -0,0 +1,204 @@ +package control + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + "sync" + + "github.com/contember/edvabe/internal/api" + "github.com/contember/edvabe/internal/runtime" +) + +// volumeNameValidator checks name constraints per the E2B OpenAPI spec. +// Names must be non-empty and match a DNS_LABEL-compatible character set +// (alphanumeric + hyphens, no leading/trailing hyphen). +func validateVolumeName(name string) bool { + if name == "" || len(name) > 63 { + return false + } + if name[0] == '-' || name[len(name)-1] == '-' { + return false + } + for _, c := range name { + if !((c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-') { + return false + } + } + return true +} + +// volumeStore is a runtime-backed volume registry. It tracks the E2B +// volume ID → logical name mapping in memory and delegates persistence +// (Docker named volumes with labels) to the runtime. A duplicate-name +// check is kept in-memory because the Docker API does not enforce name +// uniqueness on labels. +type volumeStore struct { + mu sync.Mutex + rt runtime.Runtime + byID map[string]*volumeRecord + byName map[string]*volumeRecord +} + +type volumeRecord struct { + VolumeID string + Name string +} + +func newVolumeStore(rt runtime.Runtime) *volumeStore { + return &volumeStore{ + rt: rt, + byID: make(map[string]*volumeRecord), + byName: make(map[string]*volumeRecord), + } +} + +// loadFromRuntime rebuilds the in-memory index from Docker-managed +// volumes. Called once at startup so the store survives edvabe +// restarts. No-op when the runtime is nil (test paths that don't +// need volume support). +func (s *volumeStore) loadFromRuntime(ctx context.Context) error { + if s.rt == nil { + return nil + } + volumes, err := s.rt.VolumeList(ctx) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + for _, v := range volumes { + rec := &volumeRecord{VolumeID: v.VolumeID, Name: v.Name} + s.byID[v.VolumeID] = rec + if v.Name != "" { + s.byName[v.Name] = rec + } + } + return nil +} + +type volumeCreateRequest struct { + Name string `json:"name"` +} + +type volumeResponse struct { + VolumeID string `json:"volumeID"` + Name string `json:"name"` +} + +type volumeDetailResponse struct { + VolumeID string `json:"volumeID"` + Name string `json:"name"` + Token string `json:"token"` +} + +func (s *volumeStore) list(w http.ResponseWriter, _ *http.Request) { + s.mu.Lock() + out := make([]volumeResponse, 0, len(s.byID)) + for _, rec := range s.byID { + out = append(out, volumeResponse{ + VolumeID: rec.VolumeID, + Name: rec.Name, + }) + } + s.mu.Unlock() + writeJSON(w, http.StatusOK, out) +} + +func (s *volumeStore) create(w http.ResponseWriter, r *http.Request) { + var req volumeCreateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + api.WriteError(w, http.StatusBadRequest, "invalid JSON body") + return + } + if !validateVolumeName(req.Name) { + api.WriteError(w, http.StatusBadRequest, "invalid volume name") + return + } + s.mu.Lock() + if _, exists := s.byName[req.Name]; exists { + s.mu.Unlock() + api.WriteError(w, http.StatusConflict, "volume name already exists") + return + } + s.mu.Unlock() + + volumeID := "vol_" + randomHex(8) + if err := s.rt.VolumeCreate(r.Context(), volumeID, req.Name); err != nil { + api.WriteError(w, http.StatusInternalServerError, err.Error()) + return + } + s.mu.Lock() + s.byID[volumeID] = &volumeRecord{VolumeID: volumeID, Name: req.Name} + s.byName[req.Name] = s.byID[volumeID] + s.mu.Unlock() + + writeJSON(w, http.StatusCreated, volumeResponse{ + VolumeID: volumeID, + Name: req.Name, + }) +} + +func (s *volumeStore) get(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, "/volumes/") + s.mu.Lock() + rec := s.byID[id] + if rec == nil { + rec = s.byName[id] + } + s.mu.Unlock() + if rec == nil { + api.WriteError(w, http.StatusNotFound, "volume not found") + return + } + writeJSON(w, http.StatusOK, volumeDetailResponse{ + VolumeID: rec.VolumeID, + Name: rec.Name, + Token: "edvabe-volume-token", + }) +} + +func (s *volumeStore) delete(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, "/volumes/") + s.mu.Lock() + rec := s.byID[id] + if rec == nil { + rec = s.byName[id] + } + s.mu.Unlock() + if rec == nil { + api.WriteError(w, http.StatusNotFound, "volume not found") + return + } + if err := s.rt.VolumeRemove(r.Context(), rec.VolumeID); err != nil { + if errors.Is(err, runtime.ErrVolumeInUse) || strings.Contains(err.Error(), "in use") { + api.WriteError(w, http.StatusConflict, "volume is in use") + return + } + api.WriteError(w, http.StatusInternalServerError, err.Error()) + return + } + s.mu.Lock() + delete(s.byID, rec.VolumeID) + delete(s.byName, rec.Name) + s.mu.Unlock() + w.WriteHeader(http.StatusNoContent) +} + +// resolveByName finds a volume record by its logical name. Used by +// the create-sandbox handler to validate volumeMounts references. +func (s *volumeStore) resolveByName(name string) (*volumeRecord, bool) { + s.mu.Lock() + defer s.mu.Unlock() + rec, ok := s.byName[name] + return rec, ok +} + +// dockerNameForVolumeID returns the physical Docker volume name for +// a given E2B volume ID. Used by the create-sandbox handler to build +// runtime.Mount entries. +func dockerNameForVolumeID(volumeID string) string { + return "edvabe-vol-" + volumeID +} diff --git a/internal/api/dashboard/handler.go b/internal/api/dashboard/handler.go index ed69fb7..cb0619d 100644 --- a/internal/api/dashboard/handler.go +++ b/internal/api/dashboard/handler.go @@ -86,6 +86,7 @@ type overviewSandbox struct { PausedAt string `json:"pausedAt,omitempty"` CreatedAt string `json:"createdAt"` ExpiresAt string `json:"expiresAt"` + LastActiveAt string `json:"lastActiveAt,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` CPUUsedPercent float64 `json:"cpuUsedPercent,omitempty"` MemoryUsedMiB int64 `json:"memoryUsedMiB,omitempty"` @@ -136,14 +137,15 @@ func serveOverview(opts HandlerOptions, w http.ResponseWriter, r *http.Request) for _, s := range list { entry := overviewSandbox{ - ID: s.ID, - TemplateID: s.TemplateID, - Alias: s.Alias, - State: s.State, - PauseMode: s.PauseMode, - CreatedAt: s.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"), - ExpiresAt: s.ExpiresAt.UTC().Format("2006-01-02T15:04:05Z"), - Metadata: s.Metadata, + ID: s.ID, + TemplateID: s.TemplateID, + Alias: s.Alias, + State: s.State, + PauseMode: s.PauseMode, + CreatedAt: s.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"), + ExpiresAt: s.ExpiresAt().UTC().Format("2006-01-02T15:04:05Z"), + LastActiveAt: s.LastActiveAt.UTC().Format("2006-01-02T15:04:05Z"), + Metadata: s.Metadata, } if !s.PausedAt.IsZero() { entry.PausedAt = s.PausedAt.UTC().Format("2006-01-02T15:04:05Z") diff --git a/internal/api/dispatch.go b/internal/api/dispatch.go index 46feb95..813bded 100644 --- a/internal/api/dispatch.go +++ b/internal/api/dispatch.go @@ -26,7 +26,11 @@ const ( // the headers are populated so the proxy handler doesn't have to // re-parse the host. Requests with neither go to the control plane // unchanged. -func NewRouter(control, proxy, dashboard http.Handler) http.Handler { +// +// onActivity, if non-nil, is called for every data-plane request +// (before proxying) so the sandbox manager can stamp LastActiveAt +// and reset the idle-TTL countdown. +func NewRouter(control, proxy, dashboard http.Handler, onActivity func(string)) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if dashboard != nil && strings.HasPrefix(r.URL.Path, "/dashboard") { dashboard.ServeHTTP(w, r) @@ -47,6 +51,9 @@ func NewRouter(control, proxy, dashboard http.Handler) http.Handler { control.ServeHTTP(w, r) return } + if onActivity != nil { + onActivity(id) + } proxy.ServeHTTP(w, r) }) } diff --git a/internal/api/dispatch_test.go b/internal/api/dispatch_test.go index 267fb10..6e49dce 100644 --- a/internal/api/dispatch_test.go +++ b/internal/api/dispatch_test.go @@ -10,7 +10,7 @@ func TestRouterRoutesToControlWhenNoSandbox(t *testing.T) { var hits struct{ control, proxy int } ctrl := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { hits.control++ }) prox := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { hits.proxy++ }) - h := NewRouter(ctrl, prox, nil) + h := NewRouter(ctrl, prox, nil, nil) req := httptest.NewRequest("GET", "http://localhost:3000/health", nil) h.ServeHTTP(httptest.NewRecorder(), req) @@ -24,7 +24,7 @@ func TestRouterRoutesToProxyViaHeader(t *testing.T) { var hits struct{ control, proxy int } ctrl := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { hits.control++ }) prox := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { hits.proxy++ }) - h := NewRouter(ctrl, prox, nil) + h := NewRouter(ctrl, prox, nil, nil) req := httptest.NewRequest("POST", "http://localhost:3000/process.Process/Start", nil) req.Header.Set(HeaderSandboxID, "isb_abc") @@ -40,7 +40,7 @@ func TestRouterRoutesToProxyViaHost(t *testing.T) { var seen *http.Request ctrl := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { t.Error("should not hit control") }) prox := http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { seen = r }) - h := NewRouter(ctrl, prox, nil) + h := NewRouter(ctrl, prox, nil, nil) req := httptest.NewRequest("GET", "http://49983-isb_host.localhost:3000/files?path=/etc", nil) req.Host = "49983-isb_host.localhost:3000" @@ -61,7 +61,7 @@ func TestRouterHeaderWinsOverHost(t *testing.T) { var seen *http.Request ctrl := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { t.Error("should not hit control") }) prox := http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { seen = r }) - h := NewRouter(ctrl, prox, nil) + h := NewRouter(ctrl, prox, nil, nil) req := httptest.NewRequest("GET", "http://49999-isb_host.localhost:3000/files", nil) req.Host = "49999-isb_host.localhost:3000" diff --git a/internal/api/proxy.go b/internal/api/proxy.go index bc41023..9add377 100644 --- a/internal/api/proxy.go +++ b/internal/api/proxy.go @@ -8,6 +8,7 @@ import ( "net/http/httputil" "net/url" "strconv" + "time" "github.com/contember/edvabe/internal/sandbox" ) @@ -20,6 +21,16 @@ type SandboxLookup interface { Get(id string) (*sandbox.Sandbox, error) } +// AutoResumer resumes a paused sandbox on demand. Satisfied by +// sandbox.Manager.Connect — which unpauses the container, re-inits +// the agent for stopped sandboxes, and resets the idle clock. When +// non-nil, the proxy calls it before forwarding any request to a +// paused sandbox so data-plane traffic self-heals without requiring +// an explicit SDK /connect call. +type AutoResumer interface { + Connect(ctx context.Context, id string, timeout time.Duration) (*sandbox.Sandbox, error) +} + // AgentResolver turns a sandbox ID into the host:port of its in-sandbox // agent. Satisfied by runtime.Runtime in production. type AgentResolver interface { @@ -39,7 +50,7 @@ type AgentResolver interface { // own transport layer — we don't need to touch Connection/Upgrade/etc. // explicitly. The request body is never read by edvabe; it flows // through as a stream. -func NewProxy(lookup SandboxLookup, resolver AgentResolver) http.Handler { +func NewProxy(lookup SandboxLookup, resolver AgentResolver, resumer AutoResumer) http.Handler { rp := &httputil.ReverseProxy{ // Negative flush interval = flush after every write. Required // for Connect-RPC server streams and NDJSON responses. @@ -61,7 +72,8 @@ func NewProxy(lookup SandboxLookup, resolver AgentResolver) http.Handler { return } - if _, err := lookup.Get(id); err != nil { + sb, err := lookup.Get(id) + if err != nil { if errors.Is(err, sandbox.ErrNotFound) { WriteError(w, http.StatusNotFound, fmt.Sprintf("sandbox %q not found", id)) return @@ -70,6 +82,13 @@ func NewProxy(lookup SandboxLookup, resolver AgentResolver) http.Handler { return } + if sb.State == sandbox.StatePaused && resumer != nil { + if _, err := resumer.Connect(r.Context(), id, sb.Timeout); err != nil { + WriteError(w, http.StatusServiceUnavailable, fmt.Sprintf("auto-resume %q: %v", id, err)) + return + } + } + host, port, err := resolver.AgentEndpoint(id) if err != nil { WriteError(w, http.StatusInternalServerError, diff --git a/internal/api/proxy_test.go b/internal/api/proxy_test.go index cbad1a5..63ff406 100644 --- a/internal/api/proxy_test.go +++ b/internal/api/proxy_test.go @@ -2,6 +2,7 @@ package api import ( "bufio" + "context" "encoding/json" "fmt" "net/http" @@ -77,7 +78,7 @@ func TestProxyForwardsRequest(t *testing.T) { mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{"isb_a": {ID: "isb_a"}}} rt := &fakeResolver{host: host, port: port} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("POST", front.URL+"/files?path=/etc/hostname", nil) @@ -110,7 +111,7 @@ func TestProxyForwardsRequest(t *testing.T) { func TestProxyReturns404OnUnknownSandbox(t *testing.T) { mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{}} rt := &fakeResolver{host: "127.0.0.1", port: 1} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("GET", front.URL+"/health", nil) @@ -139,7 +140,7 @@ func TestProxyReturns404OnUnknownSandbox(t *testing.T) { func TestProxyReturns400WhenHeaderMissing(t *testing.T) { mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{}} rt := &fakeResolver{host: "127.0.0.1", port: 1} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() resp, err := http.Get(front.URL + "/health") @@ -158,7 +159,7 @@ func TestProxyReturns502WhenUpstreamDown(t *testing.T) { // Point the resolver at a port nothing listens on (port 1 is the // TCPMUX well-known port; effectively always closed on dev laptops). rt := &fakeResolver{host: "127.0.0.1", port: 1} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("GET", front.URL+"/health", nil) @@ -188,7 +189,7 @@ func TestProxyPortOverride(t *testing.T) { // The resolver returns a different port (1) so the only way the // request can reach our backend is via the header override. rt := &fakeResolver{host: host, port: 1} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("GET", front.URL+"/health", nil) @@ -220,7 +221,7 @@ func TestProxyNoPortHeaderUsesDefault(t *testing.T) { mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{"isb_d": {ID: "isb_d"}}} rt := &fakeResolver{host: host, port: port} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("GET", front.URL+"/health", nil) @@ -262,7 +263,7 @@ func TestProxyStreamsResponse(t *testing.T) { mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{"isb_s": {ID: "isb_s"}}} rt := &fakeResolver{host: host, port: port} - front := httptest.NewServer(NewProxy(mgr, rt)) + front := httptest.NewServer(NewProxy(mgr, rt, nil)) defer front.Close() req, _ := http.NewRequest("GET", front.URL+"/stream", nil) @@ -306,3 +307,83 @@ func TestProxyStreamsResponse(t *testing.T) { t.Errorf("chunks too close together: t1=%v t2=%v gap=%v want>=%v", t1, t2, gap, chunkDelay-50*time.Millisecond) } } + +// fakeResumer is an AutoResumer stub that records the call and +// optionally flips the sandbox to running (simulating a successful +// resume). +type fakeResumer struct { + called bool + calledID string +} + +func (f *fakeResumer) Connect(_ context.Context, id string, _ time.Duration) (*sandbox.Sandbox, error) { + f.called = true + f.calledID = id + return nil, nil +} + +func TestAutoResumePausedSandbox(t *testing.T) { + var gotPath string + host, port, closeBackend := newBackend(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + w.WriteHeader(200) + fmt.Fprintln(w, "ok") + })) + defer closeBackend() + + sb := &sandbox.Sandbox{ID: "isb_paused", State: sandbox.StatePaused, Timeout: 5 * time.Minute} + mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{"isb_paused": sb}} + rt := &fakeResolver{host: host, port: port} + resumer := &fakeResumer{} + front := httptest.NewServer(NewProxy(mgr, rt, resumer)) + defer front.Close() + + req, _ := http.NewRequest("GET", front.URL+"/files", nil) + req.Header.Set(HeaderSandboxID, "isb_paused") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("proxy request: %v", err) + } + defer resp.Body.Close() + + if !resumer.called { + t.Error("AutoResumer.Connect was not called for paused sandbox") + } + if resumer.calledID != "isb_paused" { + t.Errorf("Connect called with id=%q, want isb_paused", resumer.calledID) + } + if resp.StatusCode != 200 { + t.Errorf("status = %d, want 200 (resume+proxy should succeed)", resp.StatusCode) + } + if gotPath != "/files" { + t.Errorf("backend path = %q, want /files", gotPath) + } +} + +func TestProxyNoResumeForRunningSandbox(t *testing.T) { + host, port, closeBackend := newBackend(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + })) + defer closeBackend() + + sb := &sandbox.Sandbox{ID: "isb_running", State: sandbox.StateRunning, Timeout: 5 * time.Minute} + mgr := &fakeLookup{byID: map[string]*sandbox.Sandbox{"isb_running": sb}} + rt := &fakeResolver{host: host, port: port} + resumer := &fakeResumer{} + front := httptest.NewServer(NewProxy(mgr, rt, resumer)) + defer front.Close() + + req, _ := http.NewRequest("GET", front.URL+"/files", nil) + req.Header.Set(HeaderSandboxID, "isb_running") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("proxy request: %v", err) + } + defer resp.Body.Close() + + if resumer.called { + t.Error("AutoResumer.Connect should not be called for running sandbox") + } +} diff --git a/internal/runtime/docker/create.go b/internal/runtime/docker/create.go index 830ad21..3e25a37 100644 --- a/internal/runtime/docker/create.go +++ b/internal/runtime/docker/create.go @@ -43,12 +43,23 @@ func (r *Runtime) Create(ctx context.Context, req runtime.CreateRequest) (*runti } var mounts []mount.Mount - for hostPath, ctrPath := range req.BindMounts { - mounts = append(mounts, mount.Mount{ - Type: mount.TypeBind, - Source: hostPath, - Target: ctrPath, - }) + for _, m := range req.Mounts { + switch m.Type { + case runtime.MountTypeBind: + mounts = append(mounts, mount.Mount{ + Type: mount.TypeBind, + Source: m.Source, + Target: m.Target, + ReadOnly: m.ReadOnly, + }) + case runtime.MountTypeVolume: + mounts = append(mounts, mount.Mount{ + Type: mount.TypeVolume, + Source: m.Source, + Target: m.Target, + ReadOnly: m.ReadOnly, + }) + } } mounts = append(mounts, r.extraBinds...) diff --git a/internal/runtime/docker/destroy.go b/internal/runtime/docker/destroy.go index 95f07d5..5f00df0 100644 --- a/internal/runtime/docker/destroy.go +++ b/internal/runtime/docker/destroy.go @@ -31,6 +31,13 @@ func (r *Runtime) Pause(ctx context.Context, sandboxID string) error { if sandboxID == "" { return fmt.Errorf("docker runtime: Pause: sandboxID is required") } + inspect, err := r.cli.ContainerInspect(ctx, sandboxID, client.ContainerInspectOptions{}) + if err != nil { + return fmt.Errorf("docker runtime: pause %q: inspect: %w", sandboxID, err) + } + if inspect.Container.State != nil && inspect.Container.State.Status == "paused" { + return nil + } if _, err := r.cli.ContainerPause(ctx, sandboxID, client.ContainerPauseOptions{}); err != nil { return fmt.Errorf("docker runtime: pause %q: %w", sandboxID, err) } @@ -43,6 +50,13 @@ func (r *Runtime) Unpause(ctx context.Context, sandboxID string) error { if sandboxID == "" { return fmt.Errorf("docker runtime: Unpause: sandboxID is required") } + inspect, err := r.cli.ContainerInspect(ctx, sandboxID, client.ContainerInspectOptions{}) + if err != nil { + return fmt.Errorf("docker runtime: unpause %q: inspect: %w", sandboxID, err) + } + if inspect.Container.State != nil && inspect.Container.State.Status == "running" { + return nil + } if _, err := r.cli.ContainerUnpause(ctx, sandboxID, client.ContainerUnpauseOptions{}); err != nil { return fmt.Errorf("docker runtime: unpause %q: %w", sandboxID, err) } diff --git a/internal/runtime/docker/volume.go b/internal/runtime/docker/volume.go new file mode 100644 index 0000000..9505e0e --- /dev/null +++ b/internal/runtime/docker/volume.go @@ -0,0 +1,171 @@ +package docker + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/mount" + "github.com/moby/moby/api/types/volume" + "github.com/moby/moby/client" + + "github.com/contember/edvabe/internal/runtime" +) + +const ( + // LabelVolumeManaged is the truthy marker on Docker volumes edvabe owns. + LabelVolumeManaged = "edvabe.volume.managed" + // LabelVolumeID stores the E2B-style vol_ ID on the Docker volume's labels. + LabelVolumeID = "edvabe.volume.id" + // LabelVolumeName stores the user-supplied logical name. + LabelVolumeName = "edvabe.volume.name" + // DockerVolumePrefix is the physical Docker volume name prefix. + DockerVolumePrefix = "edvabe-vol-" +) + +// VolumeCreate creates a managed Docker named volume with edvabe labels. +// The physical Docker volume name is DockerVolumePrefix + volumeID so +// it cannot collide with user-created Docker volumes or other edvabe +// volumes that share the same logical name. +func (r *Runtime) VolumeCreate(ctx context.Context, volumeID, name string) error { + if volumeID == "" { + return fmt.Errorf("docker runtime: VolumeCreate: volumeID is required") + } + dockerName := DockerVolumePrefix + volumeID + labels := map[string]string{ + LabelVolumeManaged: "true", + LabelVolumeID: volumeID, + LabelVolumeName: name, + } + result, err := r.cli.VolumeCreate(ctx, client.VolumeCreateOptions{ + Name: dockerName, + Labels: labels, + }) + if err != nil { + return fmt.Errorf("docker runtime: volume create %q: %w", volumeID, err) + } + + if err := r.initVolumePermissions(ctx, result.Volume.Mountpoint); err != nil { + _, _ = r.cli.VolumeRemove(ctx, dockerName, client.VolumeRemoveOptions{Force: true}) + return fmt.Errorf("docker runtime: volume init permissions %q: %w", volumeID, err) + } + return nil +} + +// initVolumePermissions chowns the volume root to the default sandbox +// user (UID 1001:GID 1001) so the user account can write to it. A new +// Docker named volume starts as root:root 0755. We run a short-lived +// helper container with the volume mounted to chown it. +func (r *Runtime) initVolumePermissions(ctx context.Context, mountpoint string) error { + helperName := "edvabe-vol-init-" + fmt.Sprintf("%d", time.Now().UnixNano()) + defer func() { + _, _ = r.cli.ContainerRemove(ctx, helperName, client.ContainerRemoveOptions{Force: true}) + }() + + cmd := []string{"chown", "-R", "1001:1001", "/vol"} + cfg := container.Config{ + Image: "busybox:latest", + Cmd: cmd, + } + hostCfg := container.HostConfig{ + Mounts: []mount.Mount{{ + Type: mount.TypeBind, + Source: mountpoint, + Target: "/vol", + }}, + AutoRemove: true, + } + created, err := r.cli.ContainerCreate(ctx, client.ContainerCreateOptions{ + Config: &cfg, + HostConfig: &hostCfg, + Name: helperName, + }) + if err != nil { + return fmt.Errorf("create helper container: %w", err) + } + if _, err := r.cli.ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil { + return fmt.Errorf("start helper container: %w", err) + } + waitResult := r.cli.ContainerWait(ctx, created.ID, client.ContainerWaitOptions{Condition: container.WaitConditionNotRunning}) + select { + case res := <-waitResult.Result: + if res.StatusCode != 0 { + return fmt.Errorf("helper container exited with status %d", res.StatusCode) + } + case err := <-waitResult.Error: + if err != nil { + return fmt.Errorf("wait helper container: %w", err) + } + case <-ctx.Done(): + return ctx.Err() + } + return nil +} + +// VolumeList returns all edvabe-managed Docker volumes. +func (r *Runtime) VolumeList(ctx context.Context) ([]runtime.VolumeInfo, error) { + filters := make(client.Filters).Add("label", LabelVolumeManaged+"=true") + result, err := r.cli.VolumeList(ctx, client.VolumeListOptions{Filters: filters}) + if err != nil { + return nil, fmt.Errorf("docker runtime: volume list: %w", err) + } + out := make([]runtime.VolumeInfo, 0, len(result.Items)) + for _, v := range result.Items { + info, ok := volumeToInfo(v) + if ok { + out = append(out, info) + } + } + return out, nil +} + +// VolumeInspect resolves a managed volume by its E2B volume ID. +func (r *Runtime) VolumeInspect(ctx context.Context, volumeID string) (*runtime.VolumeInfo, error) { + if volumeID == "" { + return nil, fmt.Errorf("docker runtime: VolumeInspect: volumeID is required") + } + dockerName := DockerVolumePrefix + volumeID + result, err := r.cli.VolumeInspect(ctx, dockerName, client.VolumeInspectOptions{}) + if err != nil { + return nil, fmt.Errorf("docker runtime: volume inspect %q: %w", volumeID, err) + } + info, ok := volumeToInfo(result.Volume) + if !ok { + return nil, fmt.Errorf("docker runtime: volume %q is not edvabe-managed", volumeID) + } + return &info, nil +} + +// VolumeRemove deletes a managed volume. Returns an error if the volume +// is in use by a running container (Docker returns a conflict error). +func (r *Runtime) VolumeRemove(ctx context.Context, volumeID string) error { + if volumeID == "" { + return fmt.Errorf("docker runtime: VolumeRemove: volumeID is required") + } + dockerName := DockerVolumePrefix + volumeID + if _, err := r.cli.VolumeRemove(ctx, dockerName, client.VolumeRemoveOptions{}); err != nil { + if strings.Contains(err.Error(), "in use") { + return fmt.Errorf("%w: %s", runtime.ErrVolumeInUse, err.Error()) + } + return fmt.Errorf("docker runtime: volume remove %q: %w", volumeID, err) + } + return nil +} + +// volumeToInfo converts a Docker volume.Volume to runtime.VolumeInfo. +// Returns ok=false if the volume lacks edvabe labels. +func volumeToInfo(v volume.Volume) (runtime.VolumeInfo, bool) { + if v.Labels == nil || v.Labels[LabelVolumeManaged] != "true" { + return runtime.VolumeInfo{}, false + } + createdAt, _ := time.Parse(time.RFC3339Nano, v.CreatedAt) + return runtime.VolumeInfo{ + VolumeID: v.Labels[LabelVolumeID], + Name: v.Labels[LabelVolumeName], + DockerName: v.Name, + CreatedAt: createdAt, + Labels: v.Labels, + }, true +} diff --git a/internal/runtime/noop/noop.go b/internal/runtime/noop/noop.go index 0265162..e046d3a 100644 --- a/internal/runtime/noop/noop.go +++ b/internal/runtime/noop/noop.go @@ -24,18 +24,26 @@ type entry struct { memoryMB int paused bool stopped bool + mounts []runtime.Mount +} + +type volumeEntry struct { + info runtime.VolumeInfo + inUse bool } type Runtime struct { mu sync.RWMutex sandboxes map[string]*entry images map[string]bool + volumes map[string]*volumeEntry } func New() *Runtime { return &Runtime{ sandboxes: make(map[string]*entry), images: make(map[string]bool), + volumes: make(map[string]*volumeEntry), } } @@ -81,6 +89,7 @@ func (r *Runtime) Create(ctx context.Context, req runtime.CreateRequest) (*runti readyCmd: req.ReadyCmd, cpuCount: req.CPUCount, memoryMB: req.MemoryMB, + mounts: copyMounts(req.Mounts), } return h, nil } @@ -304,4 +313,84 @@ func copyStrMap(m map[string]string) map[string]string { return out } +func copyMounts(mounts []runtime.Mount) []runtime.Mount { + if len(mounts) == 0 { + return nil + } + out := make([]runtime.Mount, len(mounts)) + copy(out, mounts) + return out +} + +// Mounts returns the mounts passed to Create for the named sandbox. +func (r *Runtime) Mounts(sandboxID string) []runtime.Mount { + r.mu.RLock() + defer r.mu.RUnlock() + if e, ok := r.sandboxes[sandboxID]; ok { + return copyMounts(e.mounts) + } + return nil +} + +func (r *Runtime) VolumeCreate(_ context.Context, volumeID, name string) error { + r.mu.Lock() + defer r.mu.Unlock() + if _, exists := r.volumes[volumeID]; exists { + return fmt.Errorf("noop: volume %q already exists", volumeID) + } + r.volumes[volumeID] = &volumeEntry{ + info: runtime.VolumeInfo{ + VolumeID: volumeID, + Name: name, + DockerName: "edvabe-vol-" + volumeID, + CreatedAt: time.Now(), + }, + } + return nil +} + +func (r *Runtime) VolumeList(_ context.Context) ([]runtime.VolumeInfo, error) { + r.mu.RLock() + defer r.mu.RUnlock() + out := make([]runtime.VolumeInfo, 0, len(r.volumes)) + for _, v := range r.volumes { + out = append(out, v.info) + } + return out, nil +} + +func (r *Runtime) VolumeInspect(_ context.Context, volumeID string) (*runtime.VolumeInfo, error) { + r.mu.RLock() + defer r.mu.RUnlock() + v, ok := r.volumes[volumeID] + if !ok { + return nil, fmt.Errorf("noop: volume %q not found", volumeID) + } + info := v.info + return &info, nil +} + +func (r *Runtime) VolumeRemove(_ context.Context, volumeID string) error { + r.mu.Lock() + defer r.mu.Unlock() + v, ok := r.volumes[volumeID] + if !ok { + return fmt.Errorf("noop: volume %q not found", volumeID) + } + if v.inUse { + return fmt.Errorf("noop: volume %q is in use", volumeID) + } + delete(r.volumes, volumeID) + return nil +} + +// VolumeSetInUse marks a volume as in-use (for testing conflict scenarios). +func (r *Runtime) VolumeSetInUse(volumeID string, inUse bool) { + r.mu.Lock() + defer r.mu.Unlock() + if v, ok := r.volumes[volumeID]; ok { + v.inUse = inUse + } +} + var _ runtime.Runtime = (*Runtime)(nil) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 52a2df5..a6d8018 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -8,10 +8,15 @@ package runtime import ( "context" + "errors" "io" "time" ) +// ErrVolumeInUse is returned by VolumeRemove when the volume is still +// referenced by a running container. +var ErrVolumeInUse = errors.New("volume is in use") + // Runtime is the interface every sandbox backend implements. // // Phase 1 has one implementation: Docker (internal/runtime/docker). A noop @@ -64,6 +69,20 @@ type Runtime interface { // in transitional states (dead, removing, created-but-not-started) // are filtered out. ListManaged(ctx context.Context) ([]ManagedContainer, error) + + // VolumeCreate creates a managed Docker named volume with edvabe + // labels. Returns the volume ID and Docker volume name. + VolumeCreate(ctx context.Context, volumeID, name string) error + + // VolumeList returns all edvabe-managed volumes. + VolumeList(ctx context.Context) ([]VolumeInfo, error) + + // VolumeInspect resolves a managed volume by its E2B volume ID. + VolumeInspect(ctx context.Context, volumeID string) (*VolumeInfo, error) + + // VolumeRemove deletes a managed volume. Returns an error if the + // volume is in use by a running container. + VolumeRemove(ctx context.Context, volumeID string) error } // LabelMetaPrefix is the Docker-label prefix under which the runtime @@ -103,6 +122,33 @@ type ManagedContainer struct { AgentPort int } +// Mount represents a single mount point in a sandbox container. It is +// a typed, ordered replacement for the former BindMounts map — it can +// express both Docker bind mounts (operator-controlled via +// EDVABE_EXTRA_BINDS) and Docker named volumes (E2B volumeMounts). +type Mount struct { + Type MountType + Source string // Docker volume name or host path + Target string // container path + ReadOnly bool +} + +type MountType string + +const ( + MountTypeBind MountType = "bind" + MountTypeVolume MountType = "volume" +) + +// VolumeInfo is the runtime's view of a managed Docker volume. +type VolumeInfo struct { + VolumeID string // E2B-style vol_ ID + Name string // user-supplied logical name + DockerName string // physical Docker volume name + CreatedAt time.Time + Labels map[string]string +} + // CreateRequest is the input to Runtime.Create. type CreateRequest struct { SandboxID string @@ -112,7 +158,11 @@ type CreateRequest struct { Timeout time.Duration AgentPort int AgentToken string - BindMounts map[string]string + // Mounts is the ordered list of mount points for this sandbox. + // Includes both E2B volume mounts (Docker named volumes) and + // operator-controlled bind mounts. The runtime appends its own + // extraBinds after these. + Mounts []Mount // StartCmd is the user-defined command from the template's // setStartCmd() that the edvabe-init wrapper runs alongside envd. // The runtime injects it as EDVABE_START_CMD in the container's diff --git a/internal/sandbox/labels.go b/internal/sandbox/labels.go index 2cdcc1d..4fa419d 100644 --- a/internal/sandbox/labels.go +++ b/internal/sandbox/labels.go @@ -1,7 +1,10 @@ package sandbox import ( + "encoding/json" + "strconv" "strings" + "time" "github.com/contember/edvabe/internal/runtime" ) @@ -17,6 +20,8 @@ const ( LabelTokenEnvd = "edvabe.sandbox.token.envd" LabelTokenTraffic = "edvabe.sandbox.token.traffic" LabelOnTimeout = "edvabe.sandbox.ontimeout" + LabelTimeout = "edvabe.sandbox.timeout" + LabelVolumeMounts = "edvabe.sandbox.volume.mounts" // labelImageTemplateID is the template-id label the template build // system stamps on the image itself. Used as a fallback when a @@ -30,8 +35,8 @@ const ( // label set stays minimal. func buildSandboxLabels(s *Sandbox) map[string]string { labels := map[string]string{ - LabelTemplateID: s.TemplateID, - LabelTokenEnvd: s.EnvdToken, + LabelTemplateID: s.TemplateID, + LabelTokenEnvd: s.EnvdToken, LabelTokenTraffic: s.TrafficToken, } if s.Alias != "" { @@ -40,12 +45,20 @@ func buildSandboxLabels(s *Sandbox) map[string]string { if s.OnTimeout != "" { labels[LabelOnTimeout] = string(s.OnTimeout) } + if s.Timeout > 0 { + labels[LabelTimeout] = strconv.FormatInt(int64(s.Timeout.Seconds()), 10) + } + if len(s.VolumeMounts) > 0 { + if raw, err := json.Marshal(volumeMountLabel(s.VolumeMounts)); err == nil { + labels[LabelVolumeMounts] = string(raw) + } + } return labels } // sandboxFromManaged reconstructs a Sandbox from what the runtime // reported. Fields that can't be recovered from labels or inspect -// (ExpiresAt, PausedAt) are left to the caller to default. +// (LastActiveAt, PausedAt) are left to the caller to default. func sandboxFromManaged(mc runtime.ManagedContainer) *Sandbox { state := StateRunning var pauseMode PauseMode @@ -63,7 +76,7 @@ func sandboxFromManaged(mc runtime.ManagedContainer) *Sandbox { templateID = mc.Labels[labelImageTemplateID] } - return &Sandbox{ + s := &Sandbox{ ID: mc.SandboxID, TemplateID: templateID, Alias: mc.Labels[LabelTemplateAlias], @@ -81,6 +94,48 @@ func sandboxFromManaged(mc runtime.ManagedContainer) *Sandbox { CPUCount: mc.CPUCount, MemoryMB: mc.MemoryMB, } + if raw := mc.Labels[LabelTimeout]; raw != "" { + if secs, err := strconv.ParseInt(raw, 10, 64); err == nil && secs > 0 { + s.Timeout = time.Duration(secs) * time.Second + } + } + if raw := mc.Labels[LabelVolumeMounts]; raw != "" { + s.VolumeMounts = parseVolumeMountLabel(raw) + } + return s +} + +// labelVolumeMount is the JSON shape stored in the Docker label. It +// only persists the logical name and path — the DockerName is derived +// from the volume ID at create time and not needed for rehydration +// (Docker mounts survive container stop/start/restart because the +// mount config is part of the container, not edvabe's memory). +type labelVolumeMount struct { + Name string `json:"name"` + Path string `json:"path"` +} + +func volumeMountLabel(mounts []VolumeMount) []labelVolumeMount { + out := make([]labelVolumeMount, len(mounts)) + for i, m := range mounts { + out[i] = labelVolumeMount{Name: m.Name, Path: m.Path} + } + return out +} + +func parseVolumeMountLabel(raw string) []VolumeMount { + var labels []labelVolumeMount + if err := json.Unmarshal([]byte(raw), &labels); err != nil { + return nil + } + out := make([]VolumeMount, len(labels)) + for i, l := range labels { + out[i] = VolumeMount{Name: l.Name, Path: l.Path} + } + if len(out) == 0 { + return nil + } + return out } // extractMetadataFromLabels pulls user metadata (edvabe.meta.* keys) diff --git a/internal/sandbox/manager.go b/internal/sandbox/manager.go index 2fb2080..4167113 100644 --- a/internal/sandbox/manager.go +++ b/internal/sandbox/manager.go @@ -18,7 +18,7 @@ var ( // Manager has no record of. ErrNotFound = errors.New("sandbox: not found") // ErrExpired is returned when a sandbox is still in the map but its - // ExpiresAt has lapsed — the next EnforceTimeouts pass will reap it. + // idle TTL has lapsed — the next EnforceTimeouts pass will reap it. ErrExpired = errors.New("sandbox: expired") ) @@ -111,6 +111,9 @@ type Manager struct { maxFrozen int stoppedGCAfter time.Duration + keepaliveEnabled bool + keepaliveCoalesce time.Duration + mu sync.RWMutex sandboxes map[string]*Sandbox } @@ -154,6 +157,14 @@ type Options struct { // before the reaper destroys it. Zero defaults to // DefaultStoppedGCAfter. Negative disables GC. StoppedGCAfter time.Duration + // KeepaliveEnabled controls whether data-plane traffic resets the + // idle timer. Defaults to true via main.go; set false to disable. + KeepaliveEnabled bool + // KeepaliveCoalesce is the minimum interval between two + // MarkActivity writes for the same sandbox. Prevents lock + // contention under high-frequency data-plane traffic. Defaults + // to 1s. + KeepaliveCoalesce time.Duration } // NewManager constructs a Manager. Runtime and Agent are required. @@ -182,17 +193,23 @@ func NewManager(opts Options) (*Manager, error) { if opts.StoppedGCAfter == 0 { opts.StoppedGCAfter = DefaultStoppedGCAfter } + if opts.KeepaliveCoalesce == 0 { + opts.KeepaliveCoalesce = time.Second + } + keepaliveEnabled := opts.KeepaliveEnabled return &Manager{ - rt: opts.Runtime, - ap: opts.Agent, - clock: opts.Clock, - baseImage: opts.BaseImage, - domain: opts.Domain, - resolver: opts.Resolver, - freezeDuration: opts.FreezeDuration, - maxFrozen: opts.MaxFrozen, - stoppedGCAfter: opts.StoppedGCAfter, - sandboxes: make(map[string]*Sandbox), + rt: opts.Runtime, + ap: opts.Agent, + clock: opts.Clock, + baseImage: opts.BaseImage, + domain: opts.Domain, + resolver: opts.Resolver, + freezeDuration: opts.FreezeDuration, + maxFrozen: opts.MaxFrozen, + stoppedGCAfter: opts.StoppedGCAfter, + keepaliveEnabled: keepaliveEnabled, + keepaliveCoalesce: opts.KeepaliveCoalesce, + sandboxes: make(map[string]*Sandbox), }, nil } @@ -202,12 +219,12 @@ func (m *Manager) Domain() string { return m.domain } // Rehydrate repopulates the sandbox registry from containers already on // the runtime. Called once on startup so paused / running sandboxes // survive edvabe restarts. Sandbox-level metadata (template id, tokens, -// on-timeout) is recovered from Docker labels stamped at Create time; -// Mutable fields that aren't persisted (ExpiresAt, PausedAt) get -// defaults: ExpiresAt = now + defaultTimeout for running sandboxes -// (SDK typically re-extends via SetTimeout anyway), PausedAt = now for -// paused sandboxes (the reaper will then hold them for FreezeDuration -// before demoting, which is conservative but safe). +// on-timeout, timeout) is recovered from Docker labels stamped at Create +// time; Mutable fields that aren't persisted (LastActiveAt, PausedAt) get +// defaults: LastActiveAt = now for running sandboxes (SDK typically +// re-extends via SetTimeout anyway), PausedAt = now for paused sandboxes +// (the reaper will then hold them for FreezeDuration before demoting, +// which is conservative but safe). // // Per-container failures are logged and skipped — one orphan must not // prevent edvabe from starting. Containers labeled edvabe.managed=true @@ -234,15 +251,14 @@ func (m *Manager) Rehydrate(ctx context.Context, defaultTimeout time.Duration) ( continue } s := sandboxFromManaged(mc) + if s.Timeout <= 0 { + s.Timeout = defaultTimeout + } if s.State == StatePaused { s.PausedAt = now - // Give paused sandboxes a nominal ExpiresAt in the past so - // the running-TTL reaper ignores them (only the pause-cycle - // reaper applies). Connect now skips ExpiresAt for paused - // sandboxes (see manager.Connect). - s.ExpiresAt = mc.CreatedAt + s.LastActiveAt = mc.CreatedAt } else { - s.ExpiresAt = now.Add(defaultTimeout) + s.LastActiveAt = now } if s.OnTimeout == "" { s.OnTimeout = OnTimeoutKill @@ -270,6 +286,12 @@ type CreateOptions struct { // negative number. CPUCount int MemoryMB int + // VolumeMounts is the resolved list of volume mounts to attach + // to the sandbox container. Each entry carries the logical E2B + // name/path (for API responses) and the Docker volume name (for + // the runtime). The control plane validates and resolves these + // before calling Create. + VolumeMounts []VolumeMount } // Create mints a fresh sandbox, starts its container via the runtime, @@ -315,8 +337,21 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Sandbox, err EnvdToken: envdToken, TrafficToken: trafficToken, OnTimeout: opts.OnTimeout, + Timeout: opts.Timeout, + VolumeMounts: opts.VolumeMounts, }) + // Build the runtime mount list: user volume mounts first, then + // any operator-controlled extra binds are appended by the runtime. + var rtMounts []runtime.Mount + for _, vm := range opts.VolumeMounts { + rtMounts = append(rtMounts, runtime.Mount{ + Type: runtime.MountTypeVolume, + Source: vm.DockerName, + Target: vm.Path, + }) + } + handle, err := m.rt.Create(ctx, runtime.CreateRequest{ SandboxID: id, Image: resolution.ImageTag, @@ -325,6 +360,7 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Sandbox, err Timeout: opts.Timeout, AgentPort: m.ap.Port(), AgentToken: envdToken, + Mounts: rtMounts, StartCmd: resolution.StartCmd, ReadyCmd: resolution.ReadyCmd, CPUCount: cpuCount, @@ -379,9 +415,11 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Sandbox, err Metadata: cloneMap(opts.Metadata), EnvVars: cloneMap(opts.EnvVars), CreatedAt: now, - ExpiresAt: now.Add(opts.Timeout), + Timeout: opts.Timeout, + LastActiveAt: now, CPUCount: cpuCount, MemoryMB: memoryMB, + VolumeMounts: cloneVolumeMounts(opts.VolumeMounts), } m.mu.Lock() @@ -452,13 +490,13 @@ func (m *Manager) Destroy(ctx context.Context, id string) error { return nil } -// SetTimeout resets the sandbox TTL from the current clock. Returns -// ErrNotFound if the sandbox is unknown or ErrExpired if a *running* -// sandbox's TTL already lapsed. Paused sandboxes are exempt — same -// reasoning as Manager.Connect: they live on the pause-cycle reaper -// (FreezeDuration → demote, StoppedGCAfter → destroy), not the -// running-TTL. Extending a paused sandbox's timeout is normal SDK -// flow (e.g. `sandbox.setTimeout(...)` ahead of `connect()`). +// SetTimeout resets the sandbox's idle TTL. Returns ErrNotFound if the +// sandbox is unknown or ErrExpired if a *running* sandbox's idle TTL has +// already lapsed. Paused sandboxes are exempt — same reasoning as +// Manager.Connect: they live on the pause-cycle reaper (FreezeDuration → +// demote, StoppedGCAfter → destroy), not the running-TTL. Extending a +// paused sandbox's timeout is normal SDK flow (e.g. +// `sandbox.setTimeout(...)` ahead of `connect()`). func (m *Manager) SetTimeout(id string, timeout time.Duration) error { m.mu.Lock() defer m.mu.Unlock() @@ -467,10 +505,17 @@ func (m *Manager) SetTimeout(id string, timeout time.Duration) error { return ErrNotFound } now := m.clock.Now() - if s.State != StatePaused && !s.ExpiresAt.After(now) { - return ErrExpired + if s.State != StatePaused { + ttl := s.Timeout + if ttl <= 0 { + ttl = DefaultTimeout + } + if now.Sub(s.LastActiveAt) >= ttl { + return ErrExpired + } } - s.ExpiresAt = now.Add(timeout) + s.Timeout = timeout + s.LastActiveAt = now return nil } @@ -487,17 +532,24 @@ func (m *Manager) Connect(ctx context.Context, id string, timeout time.Duration) return nil, ErrNotFound } now := m.clock.Now() - // ExpiresAt tracks the running-sandbox TTL. Paused sandboxes have - // their own lifecycle (FreezeDuration → demote, StoppedGCAfter → - // destroy), so the expiry check shouldn't block resuming them — - // otherwise any pause that outlives the original TTL is un-resumable. - if s.State != StatePaused && !s.ExpiresAt.After(now) { - m.mu.Unlock() - return nil, ErrExpired + // Paused sandboxes have their own lifecycle (FreezeDuration → + // demote, StoppedGCAfter → destroy), so the idle check shouldn't + // block resuming them — otherwise any pause that outlives the + // original TTL is un-resumable. + if s.State != StatePaused { + ttl := s.Timeout + if ttl <= 0 { + ttl = DefaultTimeout + } + if now.Sub(s.LastActiveAt) >= ttl { + m.mu.Unlock() + return nil, ErrExpired + } } priorState := s.State priorMode := s.PauseMode - s.ExpiresAt = now.Add(timeout) + s.Timeout = timeout + s.LastActiveAt = now m.mu.Unlock() if priorState != StatePaused { @@ -690,8 +742,9 @@ func (m *Manager) Snapshot(ctx context.Context, id, name string) (*SnapshotInfo, // EnforceTimeouts walks the registry and applies the three scheduled // transitions: // -// 1. Running sandboxes whose ExpiresAt has lapsed: OnTimeoutKill -// destroys them, OnTimeoutPause freezes them via runtime.Pause. +// 1. Running sandboxes whose idle TTL has lapsed (now - LastActiveAt >= +// Timeout): OnTimeoutKill destroys them, OnTimeoutPause freezes +// them via runtime.Pause. // 2. Frozen sandboxes paused longer than FreezeDuration, or any excess // beyond MaxFrozen (oldest PausedAt first): demoted via // runtime.Stop to free host RAM. @@ -732,7 +785,11 @@ func (m *Manager) EnforceTimeouts(ctx context.Context) []string { } continue } - if !s.ExpiresAt.After(now) { + ttl := s.Timeout + if ttl <= 0 { + ttl = DefaultTimeout + } + if now.Sub(s.LastActiveAt) >= ttl { if s.OnTimeout == OnTimeoutPause { toPause = append(toPause, expiredEntry{id: id, sbx: s}) } else { @@ -816,6 +873,29 @@ func (m *Manager) Run(ctx context.Context, interval time.Duration) { } } +// MarkActivity stamps the sandbox's LastActiveAt to now, resetting the +// idle-TTL countdown. Called by the dispatch layer on every data-plane +// request (any request carrying E2b-Sandbox-Id). Coalesced to at most +// one write per keepaliveCoalesce per sandbox to avoid lock contention +// under high-frequency traffic. No-op for paused sandboxes — resume +// (via Connect) resets LastActiveAt separately. +func (m *Manager) MarkActivity(id string) { + if !m.keepaliveEnabled { + return + } + m.mu.Lock() + defer m.mu.Unlock() + s, ok := m.sandboxes[id] + if !ok || s.State != StateRunning { + return + } + now := m.clock.Now() + if now.Sub(s.LastActiveAt) < m.keepaliveCoalesce { + return + } + s.LastActiveAt = now +} + func cloneMap(m map[string]string) map[string]string { if m == nil { return nil @@ -826,3 +906,12 @@ func cloneMap(m map[string]string) map[string]string { } return out } + +func cloneVolumeMounts(mounts []VolumeMount) []VolumeMount { + if len(mounts) == 0 { + return nil + } + out := make([]VolumeMount, len(mounts)) + copy(out, mounts) + return out +} diff --git a/internal/sandbox/manager_test.go b/internal/sandbox/manager_test.go index 42fc1eb..3a0913e 100644 --- a/internal/sandbox/manager_test.go +++ b/internal/sandbox/manager_test.go @@ -39,12 +39,12 @@ func (c *fakeClock) Advance(d time.Duration) { // times each callback fired so tests can assert the manager actually // handshook the agent. type stubAgent struct { - port int - pings int - inits int - readyCalls int - readyCmds []string - readyErr error + port int + pings int + inits int + readyCalls int + readyCmds []string + readyErr error } func (s *stubAgent) Name() string { return "stub" } @@ -154,8 +154,8 @@ func TestCreateGetList(t *testing.T) { if !s.CreatedAt.Equal(clk.Now()) { t.Errorf("CreatedAt = %v, want %v", s.CreatedAt, clk.Now()) } - if !s.ExpiresAt.Equal(clk.Now().Add(60 * time.Second)) { - t.Errorf("ExpiresAt = %v, want +60s", s.ExpiresAt) + if !s.ExpiresAt().Equal(clk.Now().Add(60 * time.Second)) { + t.Errorf("ExpiresAt = %v, want +60s", s.ExpiresAt()) } if ap.pings != 1 { t.Errorf("agent.Ping called %d times, want 1", ap.pings) @@ -328,8 +328,8 @@ func TestSetTimeoutExtendsPausedSandboxPastOriginalTTL(t *testing.T) { } got, _ := m.Get(s.ID) wantExpiry := clk.Now().Add(10 * time.Minute) - if !got.ExpiresAt.Equal(wantExpiry) { - t.Errorf("ExpiresAt after SetTimeout = %v, want %v", got.ExpiresAt, wantExpiry) + if !got.ExpiresAt().Equal(wantExpiry) { + t.Errorf("ExpiresAt after SetTimeout = %v, want %v", got.ExpiresAt(), wantExpiry) } } @@ -351,8 +351,8 @@ func TestConnectExtendsTimeout(t *testing.T) { if got.ID != s.ID { t.Errorf("Connect returned %q, want %q", got.ID, s.ID) } - if !got.ExpiresAt.Equal(clk.Now().Add(60 * time.Second)) { - t.Errorf("ExpiresAt = %v, want %v", got.ExpiresAt, clk.Now().Add(60*time.Second)) + if !got.ExpiresAt().Equal(clk.Now().Add(60 * time.Second)) { + t.Errorf("ExpiresAt = %v, want %v", got.ExpiresAt(), clk.Now().Add(60*time.Second)) } } @@ -386,8 +386,8 @@ func TestCreateAppliesDefaultTimeout(t *testing.T) { t.Fatalf("Create: %v", err) } want := clk.Now().Add(DefaultTimeout) - if !s.ExpiresAt.Equal(want) { - t.Errorf("ExpiresAt = %v, want %v (DefaultTimeout)", s.ExpiresAt, want) + if !s.ExpiresAt().Equal(want) { + t.Errorf("ExpiresAt = %v, want %v (DefaultTimeout)", s.ExpiresAt(), want) } } @@ -689,8 +689,8 @@ func TestConnectResumesPausedSandboxPastOriginalTTL(t *testing.T) { t.Errorf("state after Connect = %q, want running", resumed.State) } wantExpiry := clk.Now().Add(2 * time.Minute) - if !resumed.ExpiresAt.Equal(wantExpiry) { - t.Errorf("ExpiresAt = %v, want %v", resumed.ExpiresAt, wantExpiry) + if !resumed.ExpiresAt().Equal(wantExpiry) { + t.Errorf("ExpiresAt = %v, want %v", resumed.ExpiresAt(), wantExpiry) } } @@ -897,7 +897,7 @@ func TestRehydrateSkipsAlreadyKnownSandboxes(t *testing.T) { if err != nil { t.Fatalf("Create: %v", err) } - originalExpiry := s.ExpiresAt + originalExpiry := s.ExpiresAt() clk.Advance(10 * time.Minute) n, err := m.Rehydrate(ctx, 5*time.Minute) @@ -908,8 +908,8 @@ func TestRehydrateSkipsAlreadyKnownSandboxes(t *testing.T) { t.Errorf("Rehydrate should skip known sandboxes, got n=%d", n) } got, _ := m.Get(s.ID) - if !got.ExpiresAt.Equal(originalExpiry) { - t.Errorf("ExpiresAt was clobbered by Rehydrate: got %v, want %v", got.ExpiresAt, originalExpiry) + if !got.ExpiresAt().Equal(originalExpiry) { + t.Errorf("ExpiresAt was clobbered by Rehydrate: got %v, want %v", got.ExpiresAt(), originalExpiry) } } @@ -1486,7 +1486,7 @@ func TestResumePreservesTTL(t *testing.T) { if err != nil { t.Fatalf("Create: %v", err) } - originalExpiry := s.ExpiresAt + originalExpiry := s.ExpiresAt() if err := m.Pause(ctx, s.ID); err != nil { t.Fatalf("Pause: %v", err) } @@ -1498,8 +1498,8 @@ func TestResumePreservesTTL(t *testing.T) { if got.State != StateRunning { t.Errorf("State = %q, want running", got.State) } - if !got.ExpiresAt.Equal(originalExpiry) { - t.Errorf("ExpiresAt changed: %v vs %v (Resume should preserve TTL)", got.ExpiresAt, originalExpiry) + if !got.ExpiresAt().Equal(originalExpiry) { + t.Errorf("ExpiresAt changed: %v vs %v (Resume should preserve TTL)", got.ExpiresAt(), originalExpiry) } if rt.IsPaused(s.ID) { t.Error("runtime still paused after Resume") @@ -1586,3 +1586,83 @@ func TestCreateDestroysContainerWhenReadyProbeFails(t *testing.T) { } _ = rt } + +func TestCreateWithVolumeMounts(t *testing.T) { + clk := newFakeClock(time.Date(2026, 4, 15, 12, 0, 0, 0, time.UTC)) + rt := noop.New() + m, err := NewManager(Options{Runtime: rt, Agent: &stubAgent{port: 49983}, Clock: clk}) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + ctx := context.Background() + + created, err := m.Create(ctx, CreateOptions{ + TemplateID: "base", + Timeout: 5 * time.Minute, + VolumeMounts: []VolumeMount{ + {Name: "data", Path: "/mnt/data", DockerName: "edvabe-vol-vol_test123"}, + }, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + if len(created.VolumeMounts) != 1 { + t.Fatalf("expected 1 volume mount, got %d", len(created.VolumeMounts)) + } + if created.VolumeMounts[0].Name != "data" || created.VolumeMounts[0].Path != "/mnt/data" { + t.Errorf("volumeMount = %+v", created.VolumeMounts[0]) + } + + mounts := rt.Mounts(created.ID) + if len(mounts) != 1 { + t.Fatalf("runtime received %d mounts, want 1", len(mounts)) + } + if mounts[0].Source != "edvabe-vol-vol_test123" || mounts[0].Target != "/mnt/data" { + t.Errorf("runtime mount = %+v", mounts[0]) + } +} + +func TestRehydrateRestoresVolumeMounts(t *testing.T) { + clk := newFakeClock(time.Date(2026, 4, 15, 12, 0, 0, 0, time.UTC)) + rt := noop.New() + m1, err := NewManager(Options{Runtime: rt, Agent: &stubAgent{port: 49983}, Clock: clk}) + if err != nil { + t.Fatalf("NewManager #1: %v", err) + } + ctx := context.Background() + + created, err := m1.Create(ctx, CreateOptions{ + TemplateID: "base", + Timeout: 5 * time.Minute, + VolumeMounts: []VolumeMount{ + {Name: "shared", Path: "/mnt/shared", DockerName: "edvabe-vol-vol_abc"}, + }, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + + // Fresh manager, same runtime. + m2, err := NewManager(Options{Runtime: rt, Agent: &stubAgent{port: 49983}, Clock: clk}) + if err != nil { + t.Fatalf("NewManager #2: %v", err) + } + n, err := m2.Rehydrate(ctx, 10*time.Minute) + if err != nil { + t.Fatalf("Rehydrate: %v", err) + } + if n != 1 { + t.Fatalf("Rehydrate count = %d, want 1", n) + } + + got, err := m2.Get(created.ID) + if err != nil { + t.Fatalf("Get after Rehydrate: %v", err) + } + if len(got.VolumeMounts) != 1 { + t.Fatalf("expected 1 volume mount after rehydrate, got %d", len(got.VolumeMounts)) + } + if got.VolumeMounts[0].Name != "shared" || got.VolumeMounts[0].Path != "/mnt/shared" { + t.Errorf("rehydrated volumeMount = %+v", got.VolumeMounts[0]) + } +} diff --git a/internal/sandbox/sandbox.go b/internal/sandbox/sandbox.go index 972ea11..a551608 100644 --- a/internal/sandbox/sandbox.go +++ b/internal/sandbox/sandbox.go @@ -41,7 +41,7 @@ const ( ) // OnTimeoutMode controls what EnforceTimeouts does to a sandbox once -// its ExpiresAt has lapsed. The default (OnTimeoutKill) destroys the +// its idle TTL has lapsed. The default (OnTimeoutKill) destroys the // container; OnTimeoutPause freezes it via runtime.Pause and leaves it // in the registry for a later /connect to resume. Values are the same // strings the E2B SDK sends in NewSandbox.lifecycle.onTimeout. @@ -53,7 +53,7 @@ const ( ) // Sandbox is edvabe's view of one active sandbox. Fields mutated over -// the sandbox's lifetime (State, ExpiresAt) are guarded by Manager.mu. +// the sandbox's lifetime (State, LastActiveAt) are guarded by Manager.mu. // Callers that receive a *Sandbox from the Manager MUST treat it as // read-only — use Manager methods to mutate. type Sandbox struct { @@ -72,15 +72,44 @@ type Sandbox struct { // PausedAt records when the sandbox was most recently paused. Used // by the reaper to demote long-frozen containers to stopped and to // GC long-stopped containers. Zero when State != StatePaused. - PausedAt time.Time + PausedAt time.Time OnTimeout OnTimeoutMode - Metadata map[string]string - EnvVars map[string]string - CreatedAt time.Time - ExpiresAt time.Time + Metadata map[string]string + EnvVars map[string]string + CreatedAt time.Time + // Timeout is the idle-TTL set at creation time (or updated via + // SetTimeout). The sandbox is reaped when it has been idle + // (no data-plane traffic) for this duration. + Timeout time.Duration + // LastActiveAt is the timestamp of the most recent data-plane + // activity (any request routed through the proxy). Stamped by + // Manager.MarkActivity with sub-second coalescing. The idle + // deadline is LastActiveAt + Timeout. + LastActiveAt time.Time // CPUCount / MemoryMB are the resource caps applied to the // container. Zero means unlimited (Docker default). Sourced from // the template resolution and per-sandbox overrides. CPUCount int MemoryMB int + // VolumeMounts is the logical mount list resolved at create time. + // Each entry has the E2B volume name, the container path, and the + // physical Docker volume name to mount. + VolumeMounts []VolumeMount +} + +// VolumeMount is the sandbox manager's view of a volume mount. The +// logical Name and Path are what the SDK sees in API responses; the +// DockerName is the physical Docker volume name used by the runtime. +type VolumeMount struct { + Name string + Path string + DockerName string +} + +// ExpiresAt is a computed property: LastActiveAt + Timeout. It is the +// projected idle-deadline used by API responses and EnforceTimeouts. +// Updating LastActiveAt or Timeout automatically shifts this value — +// callers never write ExpiresAt directly. +func (s *Sandbox) ExpiresAt() time.Time { + return s.LastActiveAt.Add(s.Timeout) } diff --git a/internal/template/builder/executor_test.go b/internal/template/builder/executor_test.go index de858ec..62ef281 100644 --- a/internal/template/builder/executor_test.go +++ b/internal/template/builder/executor_test.go @@ -50,6 +50,18 @@ func (f *fakeRuntime) AgentEndpoint(string) (string, int, error) { func (f *fakeRuntime) ListManaged(context.Context) ([]runtime.ManagedContainer, error) { return nil, nil } +func (f *fakeRuntime) VolumeCreate(context.Context, string, string) error { + return nil +} +func (f *fakeRuntime) VolumeList(context.Context) ([]runtime.VolumeInfo, error) { + return nil, nil +} +func (f *fakeRuntime) VolumeInspect(context.Context, string) (*runtime.VolumeInfo, error) { + return nil, errors.New("not implemented") +} +func (f *fakeRuntime) VolumeRemove(context.Context, string) error { + return nil +} func (f *fakeRuntime) BuildImage(ctx context.Context, req runtime.BuildRequest) error { f.mu.Lock()