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/api/control/sandboxes.go b/internal/api/control/sandboxes.go index 4c5e1f7..dceda6a 100644 --- a/internal/api/control/sandboxes.go +++ b/internal/api/control/sandboxes.go @@ -56,6 +56,7 @@ 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 { @@ -68,6 +69,7 @@ type sandboxDetailResponse struct { Network sandboxNetworkConfig `json:"network"` AllowInternetAccess bool `json:"allowInternetAccess"` EnvVars map[string]string `json:"envVars,omitempty"` + LastActiveAt time.Time `json:"lastActiveAt"` } type sandboxLifecycle struct { @@ -549,7 +551,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 +570,7 @@ func toSandboxDetailResponse(manager sandboxManager, provider versionProvider, s Network: sandboxNetworkConfig{AllowPublicTraffic: true, AllowOut: []string{}, DenyOut: []string{}}, AllowInternetAccess: true, EnvVars: sbx.EnvVars, + LastActiveAt: sbx.LastActiveAt, } // Fall back to runtime stats for sandboxes created before this // field existed (CPUCount/MemoryMB == 0 means "unlimited" on the 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/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/sandbox/labels.go b/internal/sandbox/labels.go index 2cdcc1d..ddc6439 100644 --- a/internal/sandbox/labels.go +++ b/internal/sandbox/labels.go @@ -1,7 +1,9 @@ package sandbox import ( + "strconv" "strings" + "time" "github.com/contember/edvabe/internal/runtime" ) @@ -17,6 +19,7 @@ const ( LabelTokenEnvd = "edvabe.sandbox.token.envd" LabelTokenTraffic = "edvabe.sandbox.token.traffic" LabelOnTimeout = "edvabe.sandbox.ontimeout" + LabelTimeout = "edvabe.sandbox.timeout" // labelImageTemplateID is the template-id label the template build // system stamps on the image itself. Used as a fallback when a @@ -40,12 +43,15 @@ 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) + } 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 +69,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 +87,12 @@ 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 + } + } + return s } // extractMetadataFromLabels pulls user metadata (edvabe.meta.* keys) diff --git a/internal/sandbox/manager.go b/internal/sandbox/manager.go index 2fb2080..c61e8f5 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 @@ -315,6 +331,7 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Sandbox, err EnvdToken: envdToken, TrafficToken: trafficToken, OnTimeout: opts.OnTimeout, + Timeout: opts.Timeout, }) handle, err := m.rt.Create(ctx, runtime.CreateRequest{ @@ -379,7 +396,8 @@ 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, } @@ -452,13 +470,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 +485,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 +512,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 +722,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 +765,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 +853,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 diff --git a/internal/sandbox/manager_test.go b/internal/sandbox/manager_test.go index 42fc1eb..c91ae67 100644 --- a/internal/sandbox/manager_test.go +++ b/internal/sandbox/manager_test.go @@ -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") diff --git a/internal/sandbox/sandbox.go b/internal/sandbox/sandbox.go index 972ea11..dafb025 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 { @@ -77,10 +77,26 @@ type Sandbox struct { Metadata map[string]string EnvVars map[string]string CreatedAt time.Time - ExpiresAt 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 } + +// 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) +}