From 696447d1d05ef18bba134c4de9f9496df8a599fe Mon Sep 17 00:00:00 2001 From: "Ruihao (Ray) Zhang" Date: Tue, 10 Nov 2026 03:57:49 +0000 Subject: [PATCH 1/4] perf(v2): prewarm project Nodes and reclaim idle ones A project's Node can now be started ahead of its first session (Warm). A session arriving meanwhile waits for that same Node; if the prewarm fails, the session confirms it stopped and cold-starts once itself. A Node already up marks the runtime preparation stage as reused. Idle Nodes (no workspace owner, no Node command in flight, no caller about to own one) stop after IdleTimeout (default 60m); beyond MaxIdle (default 7) the least recently used stop first. The Demo Node is exempt, a Node still starting is not counted, and one whose stop is unconfirmed stays registered so no second incarnation starts in its directory. --- v2-server/internal/agentruntime/driver.go | 27 +- .../internal/agentruntime/market_skills.go | 5 +- v2-server/internal/agentruntime/prewarm.go | 147 +++++++ .../internal/agentruntime/prewarm_test.go | 407 ++++++++++++++++++ v2-server/internal/agentruntime/project.go | 78 +++- v2-server/internal/agentruntime/session.go | 8 +- 6 files changed, 654 insertions(+), 18 deletions(-) create mode 100644 v2-server/internal/agentruntime/prewarm.go create mode 100644 v2-server/internal/agentruntime/prewarm_test.go diff --git a/v2-server/internal/agentruntime/driver.go b/v2-server/internal/agentruntime/driver.go index f68effe57..7d12b6801 100644 --- a/v2-server/internal/agentruntime/driver.go +++ b/v2-server/internal/agentruntime/driver.go @@ -47,8 +47,21 @@ type Options struct { // RecordWorkspace remembers where a workspace's checkout is, once it is ready and before the // Agent starts, so pages can read its files later. Nil records nothing. RecordWorkspace func(ctx context.Context, workspaceID string, checkout Checkout) error + // IdleTimeout is how long a project's Node may sit idle before it is stopped; zero means + // DefaultIdleTimeout. MaxIdle caps how many idle Nodes stay up, least recently used stopped + // first; zero means DefaultMaxIdle. The bootstrapped Demo's Node is exempt from both. + IdleTimeout time.Duration + MaxIdle int } +// Idle reclaim defaults: a demo segment with its pauses stays warm, and three idle Nodes cover a +// segment's projects plus one created on stage. An idle Node costs about 7% of a core and 145 MB, +// mostly guardians left by the workspace checkouts it prepared, so the cap stays small. +const ( + DefaultIdleTimeout = 60 * time.Minute + DefaultMaxIdle = 3 +) + // Checkout is a workspace's directory on this host and the commit it was cloned at. type Checkout struct { Path string @@ -70,6 +83,10 @@ type Driver struct { serial atomic.Uint64 controlTimeout time.Duration startProject func(*projectRuntime) error + now func() time.Time // nil reads the process clock + waiting func(string) // test seam: an acquire is about to wait for a starting Node + stopReclaim chan struct{} + reclaimOnce sync.Once } // New validates deployment wiring. Processes start lazily for an already-admitted project turn. @@ -106,7 +123,12 @@ func New(opts Options) (*Driver, error) { if _, err := rand.Read(nonce[:]); err != nil { return nil, execution.ErrUnavailable } - return &Driver{opts: opts, boot: hex.EncodeToString(nonce[:]), projects: map[string]*projectRuntime{}, sessions: map[string]*agentSession{}, controlTimeout: 2 * time.Minute}, nil + if opts.IdleTimeout < 0 || opts.MaxIdle < 0 { + return nil, errors.New("invalid Agent runtime idle reclaim configuration") + } + d := &Driver{opts: opts, boot: hex.EncodeToString(nonce[:]), projects: map[string]*projectRuntime{}, sessions: map[string]*agentSession{}, controlTimeout: 2 * time.Minute, stopReclaim: make(chan struct{})} + go d.reclaimLoop() + return d, nil } // Run executes the admitted turn. Normal Append reuses its original live Agent conversation. @@ -239,6 +261,9 @@ func (d *Driver) End(ctx context.Context, sid string) error { func (d *Driver) Close(ctx context.Context) error { d.mu.Lock() d.closed = true + if d.stopReclaim != nil { + d.reclaimOnce.Do(func() { close(d.stopReclaim) }) + } projects := make([]*projectRuntime, 0, len(d.projects)) for _, p := range d.projects { projects = append(projects, p) diff --git a/v2-server/internal/agentruntime/market_skills.go b/v2-server/internal/agentruntime/market_skills.go index fc68d1aba..8c673ca10 100644 --- a/v2-server/internal/agentruntime/market_skills.go +++ b/v2-server/internal/agentruntime/market_skills.go @@ -67,9 +67,10 @@ func (d *Driver) ApplySkill(ctx context.Context, project, id, version, file, dig p.skillsMu.Lock() defer p.skillsMu.Unlock() p.mu.Lock() - active := len(p.owners) > 0 + // Idle reclaim may have retired the Node while this call waited for skillsMu. + active, retired := len(p.owners) > 0, p.fatal p.mu.Unlock() - if active { + if active || retired { return ErrOfflineSkillNode } command, outcome := "remove_plugins", "removed" diff --git a/v2-server/internal/agentruntime/prewarm.go b/v2-server/internal/agentruntime/prewarm.go new file mode 100644 index 000000000..ac78946bf --- /dev/null +++ b/v2-server/internal/agentruntime/prewarm.go @@ -0,0 +1,147 @@ +//go:build linux + +package agentruntime + +import ( + "context" + "sort" + "time" + + "github.com/wanglongan587/cloud/v2-server/internal/execution" +) + +// Warm starts a project's Node ahead of its first session and returns once it is up or has failed. +// A session arriving meanwhile waits for the same Node instead of starting another. A failure is +// only reported: the next session cleans it up and starts its own Node, as without prewarming. +// The Node then idles under the reclaim policy like any other. +func (d *Driver) Warm(ctx context.Context, id string, config Project) error { + d.mu.Lock() + if d.closed { + d.mu.Unlock() + return execution.ErrUnavailable + } + p := d.projects[id] + if p == nil { + p = d.newProjectLocked(id, config, true) + } + d.mu.Unlock() + select { + case <-p.ready: + return p.initErr + case <-ctx.Done(): + return ctx.Err() + } +} + +func (d *Driver) clock() time.Time { + if d.now != nil { + return d.now() + } + return time.Now() +} + +func (d *Driver) idleTimeout() time.Duration { + if d.opts.IdleTimeout > 0 { + return d.opts.IdleTimeout + } + return DefaultIdleTimeout +} + +func (d *Driver) maxIdle() int { + if d.opts.MaxIdle > 0 { + return d.opts.MaxIdle + } + return DefaultMaxIdle +} + +func (d *Driver) reclaimLoop() { + ticker := time.NewTicker(min(d.idleTimeout()/4, time.Minute)) + defer ticker.Stop() + for { + select { + case <-d.stopReclaim: + return + case <-ticker.C: + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + d.reclaim(ctx) + cancel() + } + } +} + +// reclaim stops idle Nodes: each one idle for longer than the idle timeout, then the least +// recently used beyond the idle cap. A Node still starting is not counted, so it is never +// stopped before it could serve. A Node whose stop is unconfirmed stays registered, so no second +// incarnation can start in its directory; the next pass, or the next session, retries the stop. +func (d *Driver) reclaim(ctx context.Context) { + d.mu.Lock() + if d.closed { + d.mu.Unlock() + return + } + type candidate struct { + p *projectRuntime + usedAt time.Time + } + var idle []candidate + var stale []*projectRuntime + for _, p := range d.projects { + select { + case <-p.ready: + default: + continue + } + if p.initErr != nil { + stale = append(stale, p) + continue + } + p.mu.Lock() + switch { + case p.fatal: + stale = append(stale, p) + case !p.warmDemo && p.idleLocked(): + idle = append(idle, candidate{p, p.usedAt}) + } + p.mu.Unlock() + } + d.mu.Unlock() + sort.Slice(idle, func(i, j int) bool { return idle[i].usedAt.Before(idle[j].usedAt) }) + cutoff := d.clock().Add(-d.idleTimeout()) + excess := len(idle) - d.maxIdle() + for i, c := range idle { + if c.p.retire(cutoff, i < excess) { + stale = append(stale, c.p) + } + } + for _, p := range stale { + if p.close(ctx) == nil { + d.dropProject(p) + } + } +} + +// idleLocked reports that nobody uses the Node: no workspace owner, no Node command in flight +// (installs, clones and skill changes all go through control) and no caller about to own a +// workspace. Caller holds p.mu. +func (p *projectRuntime) idleLocked() bool { + return len(p.owners) == 0 && len(p.controls) == 0 && p.pins == 0 +} + +// retire marks an idle Node unusable so it can be stopped: idle since before cutoff, or evicted +// over the cap. It skips a Node that became busy meanwhile, including one applying a skill change. +func (p *projectRuntime) retire(cutoff time.Time, evict bool) bool { + if !p.skillsMu.TryLock() { + return false + } + defer p.skillsMu.Unlock() + p.mu.Lock() + defer p.mu.Unlock() + if p.fatal || !p.idleLocked() || (!evict && p.usedAt.After(cutoff)) { + return false + } + // Later acquires see fatal and stop it before starting a new incarnation; skill changes see + // dead and treat the project as offline. + p.fatal = true + p.deadOnce.Do(func() { close(p.dead) }) + return true +} diff --git a/v2-server/internal/agentruntime/prewarm_test.go b/v2-server/internal/agentruntime/prewarm_test.go new file mode 100644 index 000000000..e8ed8576a --- /dev/null +++ b/v2-server/internal/agentruntime/prewarm_test.go @@ -0,0 +1,407 @@ +//go:build linux + +package agentruntime + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/wanglongan587/cloud/v2-server/internal/execution" + "github.com/wanglongan587/cloud/v2-server/internal/node" +) + +// startGate holds each Node start until the test releases it, failing it when released with an error. +type startGate struct { + entered chan string + release chan error +} + +func gateStarts(h *ipcHarness) *startGate { + g := &startGate{entered: make(chan string, 8), release: make(chan error, 8)} + start := h.driver.startProject + h.driver.startProject = func(p *projectRuntime) error { + g.entered <- p.id + if err := <-g.release; err != nil { + return err + } + return start(p) + } + return g +} + +// waitingAcquires reports each acquire that waits for a Node still starting. +func waitingAcquires(h *ipcHarness) chan string { + waiting := make(chan string, 8) + h.driver.waiting = func(id string) { waiting <- id } + return waiting +} + +// fakeClock lets a test move idle time forward. +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func useFakeClock(h *ipcHarness) *fakeClock { + c := &fakeClock{now: time.Date(2026, 11, 10, 9, 0, 0, 0, time.UTC)} + h.driver.now = func() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.now } + return c +} + +func (c *fakeClock) advance(d time.Duration) { c.mu.Lock(); c.now = c.now.Add(d); c.mu.Unlock() } + +// runtimeStage keeps the runtime stage a turn's preparation last published. +type runtimeStage struct { + mu sync.Mutex + stage *execution.PreparationStage +} + +func (r *runtimeStage) emit(_ context.Context, event execution.Event) error { + if event.Kind == execution.EventPreparation && len(event.Preparation.Stages) > 0 { + r.mu.Lock() + stage := event.Preparation.Stages[0] + r.stage = &stage + r.mu.Unlock() + } + return nil +} + +func (r *runtimeStage) reused(t *testing.T) bool { + t.Helper() + r.mu.Lock() + defer r.mu.Unlock() + if r.stage == nil || r.stage.ID != stageRuntime || r.stage.EndedAt == nil { + t.Fatalf("runtime stage = %+v", r.stage) + } + return r.stage.Reused +} + +func (h *ipcHarness) registered(id string) *projectRuntime { + h.driver.mu.Lock() + defer h.driver.mu.Unlock() + return h.driver.projects[id] +} + +func expectNoNode(t *testing.T, h *ipcHarness) { + t.Helper() + select { + case <-h.created: + t.Fatal("an unexpected Node incarnation started") + default: + } +} + +// chat runs one chat turn on project "project" and returns its runtime stage and the Node that +// served it. A nil f expects the turn to start a new incarnation. +func chat(t *testing.T, h *ipcHarness, f *fakeNodeIPC, id string) (*runtimeStage, *fakeNodeIPC) { + t.Helper() + stage := &runtimeStage{} + done := startIPCRun(h, ipcContext(t), h.session(id, id), id, stage.emit) + if f == nil { + f = receiveIPC(t, h.created) + } + f.finishTurn(t, nextIPCCommand(t, f, "start_agent_session"), id, "OK") + if got := receiveIPC(t, done); got.err != nil || got.result.Answer != "OK" { + t.Fatalf("turn %s = %+v", id, got) + } + return stage, f +} + +func endSession(t *testing.T, h *ipcHarness, f *fakeNodeIPC, id string) { + t.Helper() + ended := make(chan error, 1) + go func() { ended <- h.driver.End(ipcContext(t), id) }() + f.terminal(nextIPCCommand(t, f, "end_session")) + if err := receiveIPC(t, ended); err != nil { + t.Fatal(err) + } +} + +func TestPrewarmedNodeServesFirstSession(t *testing.T) { + h := newIPCHarness(t, nil) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err != nil { + t.Fatal(err) + } + f := receiveIPC(t, h.created) + if stage, _ := chat(t, h, f, "chat"); !stage.reused(t) { + t.Fatal("a prewarmed Node was not reported as reused") + } + expectNoNode(t, h) +} + +// #1026: a failed prewarm changes nothing for the project; its next session cold-starts. +func TestPrewarmFailureLeavesSessionToColdStart(t *testing.T) { + var installs atomic.Int32 + h := newIPCHarness(t, func(f *fakeNodeIPC, m node.Message) bool { + if m.Type == "install_plugins" && installs.Add(1) == 1 { + f.reply(m, "plugins_result", map[string]string{"kind": "plugins_failed"}) + return true + } + return false + }) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err == nil { + t.Fatal("failed prewarm reported success") + } + failed := receiveIPC(t, h.created) + if failed.stops.Load() == 0 || h.registered("project") != nil { + t.Fatal("failed prewarm was not stopped and dropped") + } + stage, f := chat(t, h, nil, "chat") + if f.id == failed.id || stage.reused(t) { + t.Fatal("session did not cold-start its own Node") + } +} + +// #1026: a session created while the Node is still warming waits for that same Node. +func TestSessionDuringPrewarmWaitsForTheSameNode(t *testing.T) { + h := newIPCHarness(t, nil) + gate, waiting := gateStarts(h), waitingAcquires(h) + warmed := make(chan error, 1) + go func() { warmed <- h.driver.Warm(ipcContext(t), "project", h.config) }() + receiveIPC(t, gate.entered) + stage := &runtimeStage{} + done := startIPCRun(h, ipcContext(t), h.session("chat", "chat"), "chat", stage.emit) + receiveIPC(t, waiting) + gate.release <- nil + if err := receiveIPC(t, warmed); err != nil { + t.Fatal(err) + } + f := receiveIPC(t, h.created) + f.finishTurn(t, nextIPCCommand(t, f, "start_agent_session"), "chat", "OK") + if got := receiveIPC(t, done); got.err != nil || got.result.Answer != "OK" { + t.Fatalf("turn = %+v", got) + } + if stage.reused(t) { + t.Fatal("a session that waited for the Node was reported as reused") + } + expectNoNode(t, h) +} + +// #1026: when the prewarm a session is waiting for fails, the session cold-starts once itself +// instead of failing the conversation. +func TestSessionDuringFailedPrewarmColdStartsOnce(t *testing.T) { + h := newIPCHarness(t, nil) + gate, waiting := gateStarts(h), waitingAcquires(h) + warmed := make(chan error, 1) + go func() { warmed <- h.driver.Warm(ipcContext(t), "project", h.config) }() + receiveIPC(t, gate.entered) + done := startIPCRun(h, ipcContext(t), h.session("chat", "chat"), "chat", discardIPCEvent) + receiveIPC(t, waiting) + gate.release <- errors.New("prewarm failed") + if err := receiveIPC(t, warmed); err == nil { + t.Fatal("failed prewarm reported success") + } + receiveIPC(t, gate.entered) // the session's own cold start + gate.release <- nil + f := receiveIPC(t, h.created) + f.finishTurn(t, nextIPCCommand(t, f, "start_agent_session"), "chat", "OK") + if got := receiveIPC(t, done); got.err != nil || got.result.Answer != "OK" { + t.Fatalf("turn = %+v", got) + } +} + +// A cold start the session asked for itself still fails as before: there is no second attempt. +func TestSessionOwnColdStartFailureIsNotRetried(t *testing.T) { + h := newIPCHarness(t, nil) + gate := gateStarts(h) + done := startIPCRun(h, ipcContext(t), h.session("chat", "chat"), "chat", discardIPCEvent) + receiveIPC(t, gate.entered) + gate.release <- errors.New("start failed") + if got := receiveIPC(t, done); !errors.Is(got.err, execution.ErrUnavailable) { + t.Fatalf("turn = %+v", got) + } + select { + case <-gate.entered: + t.Fatal("a session's own failed start was retried") + default: + } +} + +// #1026: after an idle Node is reclaimed, the next session starts a new one. +func TestReclaimedNodeIsReplacedForNextSession(t *testing.T) { + h := newIPCHarness(t, nil) + clock := useFakeClock(h) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err != nil { + t.Fatal(err) + } + old := receiveIPC(t, h.created) + clock.advance(DefaultIdleTimeout - time.Second) + h.driver.reclaim(ipcContext(t)) + if old.stops.Load() != 0 { + t.Fatal("Node reclaimed before its idle timeout") + } + clock.advance(2 * time.Second) + h.driver.reclaim(ipcContext(t)) + if old.stops.Load() == 0 || h.registered("project") != nil { + t.Fatal("idle Node was not reclaimed") + } + stage, f := chat(t, h, nil, "chat") + if f.id == old.id || stage.reused(t) { + t.Fatal("session did not start a new Node after reclaim") + } +} + +func TestReclaimKeepsNodesInUse(t *testing.T) { + h := newIPCHarness(t, nil) + h.driver.controlTimeout = 5 * time.Second + clock := useFakeClock(h) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err != nil { + t.Fatal(err) + } + f := receiveIPC(t, h.created) + reclaimAfterTimeout := func() { + clock.advance(DefaultIdleTimeout + time.Minute) + h.driver.reclaim(ipcContext(t)) + } + + // A live conversation owns its workspace for as long as it lasts. + chat(t, h, f, "chat") + reclaimAfterTimeout() + if f.stops.Load() != 0 { + t.Fatal("reclaimed a Node with a live session") + } + endSession(t, h, f, "chat") + h.driver.reclaim(ipcContext(t)) + if f.stops.Load() != 0 { + t.Fatal("idle time did not restart when the session ended") + } + + // A caller that found the Node but does not own a workspace yet. + p := h.registered("project") + if _, _, err := h.driver.acquire(ipcContext(t), "project", h.config, true); err != nil { + t.Fatal(err) + } + reclaimAfterTimeout() + if f.stops.Load() != 0 { + t.Fatal("reclaimed a pinned Node") + } + p.unpin() + + // A Node command in flight, such as a skill change. + replied := make(chan error, 1) + go func() { + _, err := p.control(ipcContext(t), "skill", "remove_plugins", "plugins_result", map[string]any{}) + replied <- err + }() + command := nextIPCCommand(t, f, "remove_plugins") + reclaimAfterTimeout() + if f.stops.Load() != 0 { + t.Fatal("reclaimed a Node with a command in flight") + } + f.reply(command, "plugins_result", map[string]string{"kind": "plugins_completed"}) + if err := receiveIPC(t, replied); err != nil { + t.Fatal(err) + } + reclaimAfterTimeout() + if f.stops.Load() == 0 { + t.Fatal("Node idle again was not reclaimed") + } +} + +func TestReclaimKeepsTheDemoNode(t *testing.T) { + h := newIPCHarness(t, nil) + clock := useFakeClock(h) + h.driver.WarmDemo(ipcContext(t), "project", h.config) + f := receiveIPC(t, h.created) + p := h.registered("project") + p.mu.Lock() + spare := p.spare + p.mu.Unlock() + receiveIPC(t, spare.ready) + clock.advance(24 * time.Hour) + h.driver.reclaim(ipcContext(t)) + if f.stops.Load() != 0 || h.registered("project") != p { + t.Fatal("reclaimed the Demo Node") + } +} + +func TestReclaimStopsLeastRecentlyUsedOverTheCap(t *testing.T) { + h := newIPCHarness(t, nil) + h.driver.opts.MaxIdle = 2 + clock := useFakeClock(h) + var nodes []*fakeNodeIPC + for _, id := range []string{"a", "b", "c"} { + if err := h.driver.Warm(ipcContext(t), id, h.config); err != nil { + t.Fatal(err) + } + nodes = append(nodes, receiveIPC(t, h.created)) + clock.advance(time.Minute) + } + h.driver.reclaim(ipcContext(t)) + if nodes[0].stops.Load() == 0 || h.registered("a") != nil { + t.Fatal("least recently used idle Node was not stopped over the cap") + } + if nodes[1].stops.Load() != 0 || nodes[2].stops.Load() != 0 { + t.Fatal("stopped Nodes within the cap") + } +} + +func TestReclaimWithUnconfirmedStopKeepsTheNodeRegistered(t *testing.T) { + h := newIPCHarness(t, nil) + clock := useFakeClock(h) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err != nil { + t.Fatal(err) + } + old := receiveIPC(t, h.created) + old.stopOK.Store(false) + clock.advance(DefaultIdleTimeout + time.Minute) + h.driver.reclaim(ipcContext(t)) + if old.stops.Load() == 0 || h.registered("project") == nil { + t.Fatal("a Node whose stop is unconfirmed was dropped") + } + // No second incarnation may start in its directory until the stop is confirmed. + got := receiveIPC(t, startIPCRun(h, ipcContext(t), h.session("blocked", "blocked"), "blocked", discardIPCEvent)) + if !errors.Is(got.err, execution.ErrUnavailable) { + t.Fatalf("turn on an unconfirmed Node = %+v", got) + } + expectNoNode(t, h) + old.stopOK.Store(true) + h.driver.reclaim(ipcContext(t)) + if h.registered("project") != nil { + t.Fatal("the next pass did not retry the stop") + } + if _, f := chat(t, h, nil, "chat"); f.id == old.id { + t.Fatal("session reused the stopped Node") + } +} + +func TestIdleNodeCrashIsReplacedForNextSession(t *testing.T) { + h := newIPCHarness(t, nil) + if err := h.driver.Warm(ipcContext(t), "project", h.config); err != nil { + t.Fatal(err) + } + old := receiveIPC(t, h.created) + p := h.registered("project") + old.Close() + receiveIPC(t, p.dead) + if _, f := chat(t, h, nil, "chat"); f.id == old.id || old.stops.Load() == 0 { + t.Fatal("crashed idle Node was not stopped and replaced") + } +} + +func TestCloseDuringPrewarmStopsTheNode(t *testing.T) { + h := newIPCHarness(t, nil) + gate := gateStarts(h) + warmed := make(chan error, 1) + go func() { warmed <- h.driver.Warm(ipcContext(t), "project", h.config) }() + receiveIPC(t, gate.entered) + closed := make(chan error, 1) + go func() { closed <- h.driver.Close(ipcContext(t)) }() + gate.release <- nil + f := receiveIPC(t, h.created) + if err := receiveIPC(t, closed); err != nil { + t.Fatal(err) + } + receiveIPC(t, warmed) + if f.stops.Load() == 0 { + t.Fatal("Close left the prewarmed Node running") + } + if err := h.driver.Warm(ipcContext(t), "other", h.config); !errors.Is(err, execution.ErrUnavailable) { + t.Fatalf("Warm after Close = %v", err) + } + expectNoNode(t, h) +} diff --git a/v2-server/internal/agentruntime/project.go b/v2-server/internal/agentruntime/project.go index b60fbcf08..e083d0894 100644 --- a/v2-server/internal/agentruntime/project.go +++ b/v2-server/internal/agentruntime/project.go @@ -53,6 +53,9 @@ type nodeClient interface { // no per-turn goroutine may call Client.Receive or close the project's shared transport. type projectRuntime struct { warmDemo bool + prewarm bool // Started by Warm, not by a session. + pins int // Callers between finding this Node and owning a workspace. + usedAt time.Time // When it last stopped serving someone; idle reclaim counts from here. spare *checkout // Unclaimed Demo checkout, never contains user skills. skillsMu sync.Mutex driver *Driver @@ -107,48 +110,91 @@ func (d *Driver) dropProject(p *projectRuntime) { } func (d *Driver) project(ctx context.Context, id string, config Project) (*projectRuntime, error) { + p, _, err := d.acquire(ctx, id, config, false) + return p, err +} + +// newProjectLocked registers a Node incarnation for id and starts it. Caller holds d.mu. +func (d *Driver) newProjectLocked(id string, config Project, prewarm bool) *projectRuntime { + lifetime, cancel := context.WithCancel(context.Background()) + p := &projectRuntime{driver: d, id: id, root: filepath.Join(d.opts.Root, digest(id)[:16]), configuration: config, prewarm: prewarm, usedAt: d.clock(), + ready: make(chan struct{}), ctx: lifetime, cancel: cancel, loopDone: make(chan struct{}), dead: make(chan struct{}), + controls: map[string]*controlWait{}, agents: map[string]*agentSession{}, checkouts: map[string]*checkout{}, installs: map[string]*installation{}, owners: map[string]string{}, acknowledged: map[string]uint64{}} + d.projects[id] = p + go p.initialize() + return p +} + +// acquire returns the project's live Node, starting one when none is registered. With pin, idle +// reclaim leaves the Node alone until the caller calls unpin, which a session does once it owns +// its workspace. reused reports that the Node was already up, so the caller waited for nothing. +func (d *Driver) acquire(ctx context.Context, id string, config Project, pin bool) (*projectRuntime, bool, error) { + fellBack := false for { d.mu.Lock() if d.closed { d.mu.Unlock() - return nil, execution.ErrUnavailable + return nil, false, execution.ErrUnavailable } p := d.projects[id] + reused := p != nil if p == nil { - lifetime, cancel := context.WithCancel(context.Background()) - p = &projectRuntime{driver: d, id: id, root: filepath.Join(d.opts.Root, digest(id)[:16]), configuration: config, - ready: make(chan struct{}), ctx: lifetime, cancel: cancel, loopDone: make(chan struct{}), dead: make(chan struct{}), - controls: map[string]*controlWait{}, agents: map[string]*agentSession{}, checkouts: map[string]*checkout{}, installs: map[string]*installation{}, owners: map[string]string{}, acknowledged: map[string]uint64{}} - d.projects[id] = p - go p.initialize() + p = d.newProjectLocked(id, config, false) } d.mu.Unlock() select { case <-p.ready: - case <-ctx.Done(): - return nil, ctx.Err() + default: + reused = false + if d.waiting != nil { + d.waiting(id) + } + select { + case <-p.ready: + case <-ctx.Done(): + return nil, false, ctx.Err() + } } if p.initErr != nil { // The failed initializer has already attempted cleanup. A later caller may retry // cleanup, but must never spawn another incarnation without stop confirmation. - if p.close(ctx) == nil { - d.dropProject(p) + if p.close(ctx) != nil { + return nil, false, execution.ErrUnavailable } - return nil, execution.ErrUnavailable + d.dropProject(p) + // A failed prewarm was nobody's request: start this caller's own Node once, the + // cold start it would have had without prewarming. + if p.prewarm && !fellBack { + fellBack = true + continue + } + return nil, false, execution.ErrUnavailable } p.mu.Lock() fatal := p.fatal + if !fatal && pin { + p.pins++ + } p.mu.Unlock() if !fatal { - return p, nil + return p, reused, nil } + // Dead or retired by idle reclaim: stop it for good, then start a fresh incarnation. if p.close(ctx) != nil { - return nil, execution.ErrUnavailable + return nil, false, execution.ErrUnavailable } d.dropProject(p) } } +// unpin releases an acquire pin; the Node counts as used now. +func (p *projectRuntime) unpin() { + p.mu.Lock() + p.pins-- + p.usedAt = p.driver.clock() + p.mu.Unlock() +} + func (p *projectRuntime) initialize() { start := p.driver.startProject if start == nil { @@ -168,6 +214,10 @@ func (p *projectRuntime) initialize() { p.unlockClose() } cancel() + } else { + p.mu.Lock() + p.usedAt = p.driver.clock() + p.mu.Unlock() } close(p.ready) } diff --git a/v2-server/internal/agentruntime/session.go b/v2-server/internal/agentruntime/session.go index 4c84c5610..d28e71985 100644 --- a/v2-server/internal/agentruntime/session.go +++ b/v2-server/internal/agentruntime/session.go @@ -95,12 +95,17 @@ func (d *Driver) session(ctx context.Context, scope execution.Scope, session exe prepare, cancel := context.WithTimeout(context.Background(), 3*time.Minute) defer cancel() timing.enter(stageRuntime, time.Now()) - p, err := d.project(prepare, session.Target.ProjectID, config) + p, reused, err := d.acquire(prepare, session.Target.ProjectID, config, true) a.project = p if err != nil { a.prepareErr = err return } + // Owning a workspace (or failing) ends the pin; until then idle reclaim must wait. + defer p.unpin() + if reused { + timing.reused() + } if p.configuration.MirrorPath != config.MirrorPath { a.prepareErr = execution.ErrUnavailable return @@ -252,6 +257,7 @@ func (a *agentSession) markStopped() { a.project.mu.Lock() if a.project.owners[a.frozen.WorkspaceID] == a.frozen.ID { delete(a.project.owners, a.frozen.WorkspaceID) + a.project.usedAt = a.project.driver.clock() } a.project.mu.Unlock() } From eebadd08a9e993189a199c4ae100f32ccdc9d654 Mon Sep 17 00:00:00 2001 From: "Ruihao (Ray) Zhang" Date: Tue, 10 Nov 2026 03:57:49 +0000 Subject: [PATCH 2/4] feat(v2): list projects by latest activity for prewarming RecentProjects orders projects across tenants by their latest chat conversation, or their creation when they have none. --- v2-server/internal/store/projects.go | 29 +++++++++++++++++++++++ v2-server/internal/store/projects_test.go | 21 ++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/v2-server/internal/store/projects.go b/v2-server/internal/store/projects.go index a74924723..4c6a4b4ef 100644 --- a/v2-server/internal/store/projects.go +++ b/v2-server/internal/store/projects.go @@ -93,3 +93,32 @@ func (s *Store) ListProjects(ctx context.Context, scope execution.Scope) ([]Proj } return projects, rows.Err() } + +// RecentProjects lists up to limit projects across tenants, most recently active first: a +// project's activity is its latest chat conversation, or its creation when it has none. It is +// backend-only (it picks which Nodes to start ahead of use) and never reaches an HTTP response. +func (s *Store) RecentProjects(ctx context.Context, limit int) ([]Project, error) { + rows, err := s.pool.Query(ctx, ` + SELECT p.data FROM v2_projects p + LEFT JOIN LATERAL (SELECT max(c.created_at) AS at FROM v2_chat_conversations c + WHERE c.tenant_id=p.tenant_id AND c.project_id=p.id) c ON true + ORDER BY greatest(c.at, (p.data->>'CreatedAt')::timestamptz) DESC NULLS LAST, p.id + LIMIT $1`, limit) + if err != nil { + return nil, err + } + defer rows.Close() + projects := []Project{} + for rows.Next() { + var raw []byte + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var project Project + if err := json.Unmarshal(raw, &project); err != nil { + return nil, err + } + projects = append(projects, project) + } + return projects, rows.Err() +} diff --git a/v2-server/internal/store/projects_test.go b/v2-server/internal/store/projects_test.go index 92a479864..aa3f0019d 100644 --- a/v2-server/internal/store/projects_test.go +++ b/v2-server/internal/store/projects_test.go @@ -3,6 +3,7 @@ package store_test import ( "errors" "testing" + "time" "github.com/wanglongan587/cloud/v2-server/internal/execution" "github.com/wanglongan587/cloud/v2-server/internal/store" @@ -48,3 +49,23 @@ func TestProjectsSharedOnlyWithinCurrentTenantMembership(t *testing.T) { t.Fatalf("revoked membership=%v", err) } } + +func TestRecentProjectsOrdersByLatestConversationThenCreation(t *testing.T) { + f := executionDatabase(t) + ctx := t.Context() + base := time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC) + for i, id := range []string{"old", "middle", "new"} { + if _, err := f.s.CreateProject(ctx, f.scope, store.Project{ID: id, Name: id, RepositoryURL: "https://example.com/" + id + ".git", Commit: "fixed", MirrorPath: "/private/" + id + ".git", CreatedAt: base.Add(time.Duration(i) * time.Hour)}); err != nil { + t.Fatal(err) + } + } + // A conversation makes the oldest project the most recently active one. + if _, err := f.admin.Exec(ctx, `INSERT INTO v2_chat_conversations(tenant_id,user_id,create_key,project_id,executor_id,title,created_at) + VALUES($1,$2,'k','old','opencode','t',$3)`, f.scope.TenantID, f.scope.UserID, base.Add(5*time.Hour)); err != nil { + t.Fatal(err) + } + got, err := f.s.RecentProjects(ctx, 2) + if err != nil || len(got) != 2 || got[0].ID != "old" || got[1].ID != "new" || got[0].MirrorPath != "/private/old.git" { + t.Fatalf("recent=%+v err=%v", got, err) + } +} From a425b6674a291fa887d53ddb3db98361e886588c Mon Sep 17 00:00:00 2001 From: "Ruihao (Ray) Zhang" Date: Tue, 10 Nov 2026 03:57:49 +0000 Subject: [PATCH 3/4] feat(v2): announce projects created through the API server.Options.ProjectCreated is told about each stored project, so main can prewarm its Node without the creation request waiting. --- v2-server/internal/server/projects.go | 3 +++ v2-server/internal/server/projects_test.go | 7 ++++++- v2-server/internal/server/server.go | 3 +++ 3 files changed, 12 insertions(+), 1 deletion(-) diff --git a/v2-server/internal/server/projects.go b/v2-server/internal/server/projects.go index 3e04c9d61..764e633f6 100644 --- a/v2-server/internal/server/projects.go +++ b/v2-server/internal/server/projects.go @@ -59,6 +59,9 @@ func (h *handler) projects(w http.ResponseWriter, r *http.Request, user store.Us projectFault(w, err) return } + if h.opts.ProjectCreated != nil { + h.opts.ProjectCreated(project) + } writeJSON(w, http.StatusCreated, projectResponse(scope, project)) return } diff --git a/v2-server/internal/server/projects_test.go b/v2-server/internal/server/projects_test.go index 059152cf2..b3f5fea67 100644 --- a/v2-server/internal/server/projects_test.go +++ b/v2-server/internal/server/projects_test.go @@ -40,7 +40,8 @@ func TestProjectRepositoryLifecycle(t *testing.T) { t.Fatal(err) } storetest.SignIn(t, s, "project-owner", "Owner", store.RoleOwner) - serverOptions := server.Options{SessionTTL: time.Hour, Projects: projects} + var createdProjects []store.Project + serverOptions := server.Options{SessionTTL: time.Hour, Projects: projects, ProjectCreated: func(p store.Project) { createdProjects = append(createdProjects, p) }} srv := httptest.NewServer(server.New(s, serverOptions)) defer func() { srv.Close() }() jar, _ := cookiejar.New(nil) @@ -130,6 +131,10 @@ func TestProjectRepositoryLifecycle(t *testing.T) { if stored.MirrorPath != filepath.Join(root, id+".git") || stored.SnapshotRef != projectrepos.SnapshotRef(id, commit) { t.Fatalf("stored repository=%+v", stored) } + // Only the stored project is announced (to prewarm its Node); failed creations are not. + if len(createdProjects) != 1 || createdProjects[0] != stored { + t.Fatalf("announced projects=%+v want %+v", createdProjects, stored) + } if got := projectGit(t, stored.MirrorPath, "rev-parse", stored.SnapshotRef); got != commit { t.Fatalf("snapshot=%q want=%q", got, commit) } diff --git a/v2-server/internal/server/server.go b/v2-server/internal/server/server.go index 0d6856e6c..a28ce2083 100644 --- a/v2-server/internal/server/server.go +++ b/v2-server/internal/server/server.go @@ -65,6 +65,9 @@ type Options struct { SessionTTL time.Duration // Projects prepares fixed repository mirrors; nil leaves project operations unavailable. Projects *projectrepos.Service + // ProjectCreated is told about each project created through the API, after it is stored and + // before the response is written; it must return at once. Nil does nothing. + ProjectCreated func(store.Project) // Execution is the shared backend for page-owned conversations and runs. // Nil uses an unavailable Stub; an explicitly supplied backend is kept unchanged. Execution execution.Service From 5b14da5e6e2efa601bc0b2aee369d853ade935ef Mon Sep 17 00:00:00 2001 From: "Ruihao (Ray) Zhang" Date: Tue, 10 Nov 2026 03:57:49 +0000 Subject: [PATCH 4/4] perf(v2): prewarm Nodes at startup and after project creation Startup warms the Demo first, then the most recently active projects up to V2_NODE_IDLE_MAX, one at a time; a created project is warmed in the background. V2_NODE_IDLE_TIMEOUT and V2_NODE_IDLE_MAX tune idle reclaim. The README describes prewarming and the reclaim policy. --- v2-server/README.md | 16 +++++++- v2-server/main.go | 11 ++++- v2-server/prewarm.go | 84 +++++++++++++++++++++++++++++++++++++++ v2-server/prewarm_test.go | 29 ++++++++++++++ 4 files changed, 137 insertions(+), 3 deletions(-) create mode 100644 v2-server/prewarm.go create mode 100644 v2-server/prewarm_test.go diff --git a/v2-server/README.md b/v2-server/README.md index 4fe577f75..f9d7ca335 100644 --- a/v2-server/README.md +++ b/v2-server/README.md @@ -35,6 +35,8 @@ go run . # 监听 127.0.0.1:8090 | `V2_MARKET_SOCKS_PROXY` | 未设置 | 自定义来源启用代理时使用的部署级 SOCKS5 地址；Git 与 Skill 包下载都使用，客户端只把已校验的公网 IP 交给代理 | | `V2_MARKET_CREDENTIAL_KEY` | 未设置 | S3 来源凭据的 AES-256-GCM 密钥（Base64 编码的 32 字节）；启用 S3 来源时必须持久提供 | | `V2_NODE_ASSETS_DIR` | 未设置 | `task v2:node:prepare` 生成的固定 Node、host、guardian、Deno、OpenCode 和 Claude Code 资源；未设置时只启用纯模型路径 | +| `V2_NODE_IDLE_TIMEOUT` | `60m` | 项目 Node 空闲多久后停止；必须为正 duration。空闲指没有进行中的会话和 Node 命令，见下文「项目 Node 的预热与回收」 | +| `V2_NODE_IDLE_MAX` | `3` | 最多保留几个空闲的项目 Node，超出时先停最久没用的；必须为正整数。Demo 项目的 Node 不计入 | | `V2_MODEL_BASE_URL` / `V2_MODEL_API_KEY` / `V2_MODEL_NAME` | 未设置 | 可选的一组 HTTPS OpenAI 兼容 provider；必须同时设置，OpenCode 与纯模型共用，清单只公开模型名称与版本 | | `V2_ANTHROPIC_MODEL_BASE_URL` / `V2_ANTHROPIC_MODEL_API_KEY` / `V2_ANTHROPIC_MODEL_NAME` | 未设置 | 可选的一组 HTTPS Anthropic 兼容 provider（Messages API，base URL 写到 `/v1`，如 `https://api.anthropic.com/v1`）；必须同时提供地址、凭据和模型名，只供 Claude Code 使用。使用 Bearer 网关时以 `V2_ANTHROPIC_MODEL_AUTH_TOKEN` 代替 `V2_ANTHROPIC_MODEL_API_KEY`，二者不能同时配置；Claude Code 的 base URL 若以 `/compatible` 结尾，这里填 `/compatible/v1` | @@ -471,6 +473,18 @@ V2_NODE_ASSETS_DIR=/absolute/path/to/assets TEST_DATABASE_URL='postgres://ora:or 用 debug 构建的 Node 首次安装约需一分钟；安装结果留在 `V2_DATA_DIR` 的项目 Node 目录里， 之后重启服务再用只需约一秒。演示前先在每个演示项目里跑一轮 Claude Code，可避开这次等待。 +### 项目 Node 的预热与回收 + +每个项目一个 Node，项目下的对话、工作流和任务共用它，各自只有独立的 checkout。为省掉新对话等 Node 启动（SSE `preparation` 的 `runtime` 段，约 3–4 秒），Node 在有人用之前就起好： + +- **预热**：项目经 API 创建成功后，后台为它起 Node，创建请求不等待。服务启动时先起 Demo（另备一份 checkout），再按最近活动（最新对话的创建时间，没有对话时取项目创建时间）依次起最多 `V2_NODE_IDLE_MAX` 个项目的 Node，一次一个，避免占满演示机的 CPU。 +- **复用**：会话来时 Node 已经在跑，`runtime` 段标为 `reused`；Node 还在启动，会话等同一个 Node，不另起。 +- **失败**：预热失败只记日志，不影响项目创建。会话正在等的预热失败时，会话先确认失败的 Node 已经停止，再自己冷启动一次；会话自己发起的冷启动失败照旧不重试。 +- **回收**：没有会话占用工作目录、没有进行中的 Node 命令（插件安装、clone、Skill 变更）、也没有调用方正要占用时，Node 才算空闲。空闲超过 `V2_NODE_IDLE_TIMEOUT` 的停止；空闲的超过 `V2_NODE_IDLE_MAX` 个时，先停最久没用的。还在启动的不计入，Demo 不回收。停止得不到确认时 Node 保持登记，同一目录不会起第二个 Node，下一轮回收或下一个会话再试。回收后不自动重起，下一个会话冷启动。 +- 对话会话在对话删除或服务重启前一直存在，所以有未删对话的项目，Node 不会被回收；回收主要作用于预热后没人用的、对话都删了的、只跑过工作流或任务的项目。一个空闲 Node 实测约占 7% 个 CPU 核和 145 MB 内存，大头是它准备过的工作目录留下的 guardian 进程（每次准备 16 个，Node 停止时才退出），所以上限取得较小。 + +正式 Cloud 按运行准备沙盒，没有这套按项目常驻、按空闲回收的 Node。 + End 和重启不自动删除目录。停止服务后 `go run . -cleanup` 只删除确认停止后的登记 Node 与对话目录， 保留项目 mirror 和数据库历史；命令拒绝与持有同一数据目录的服务并行运行。 外部 provider 的显式验收入口与配置见执行服务包 README；默认 skip 不代表外部 provider 验收通过。 @@ -613,7 +627,7 @@ npm run e2e -- e2e/workflow-runs.spec.ts e2e/workflow-containers.spec.ts e2e/sch 与 Desktop 的差异： - Desktop 从 Agent 插件声明中发现 Skill 目录，复制到工作区并写 ownership marker；v2 固定使用 `.opencode/skills`、`.claude/skills`，以 symlink 指向该项目的 Node 内部安装目录，因此依赖当前 Node 的目录布局。 -- Desktop 会先让 Agent 停下，再在会话进行中应用变化；v2 只在项目 Node 已在线且项目空闲时安装或移除。新选择要等该项目有一个会话结束后才生效，服务启动后的第一个会话也拿不到尚未应用的 Skill。 +- Desktop 会先让 Agent 停下，再在会话进行中应用变化；v2 只在项目 Node 已在线且项目空闲时安装或移除。新选择要等该项目有一个会话结束后才生效；Node 预热上线后、第一个会话开始前项目是空闲的，这时也会应用。Node 被回收后项目离线，选择等它下次上线再应用。 测试真实 Node 时设置 `V2_NODE_ASSETS_DIR`、`V2_TEST_SKILL_ARCHIVE`（官方 Matt Pocock Skills 1.3.1）和可选的 `V2_TEST_SKILL_UPDATE_ARCHIVE`（1.3.2），在 Linux 原生、路径足够短的用户目录下运行 `go test -run TestRealMarketSkillNode ./internal/agentruntime`。该测试用真实 OpenCode 读取插件 Skill 和工作流所选个人 Skill，并验证下一节点清除两者；模型响应仍为 fixture，不验证真实模型的指令遵循能力。 diff --git a/v2-server/main.go b/v2-server/main.go index 8fc0a86c7..44d4e3f66 100644 --- a/v2-server/main.go +++ b/v2-server/main.go @@ -122,6 +122,10 @@ func run(addr, dbURL string, cleanup, fixture, testFixtures bool) (resultErr err if fixture || testFixtures { models = executionruntime.WithFixtureModels(models) } + idleTimeout, idleMax, e := nodeIdlePolicy(os.Getenv) + if e != nil { + return e + } var driver *agentruntime.Driver if assets := os.Getenv("V2_NODE_ASSETS_DIR"); assets != "" && len(models) > 0 { gitDir, e := exec.CommandContext(ctx, "git", "--exec-path").Output() @@ -132,7 +136,7 @@ func run(addr, dbURL string, cleanup, fixture, testFixtures bool) (resultErr err for id, choice := range models { bindings[id] = choice.Binding } - driver, e = agentruntime.New(agentruntime.Options{Root: filepath.Join(base, "nodes"), AssetsDirectory: assets, GitBackend: filepath.Join(strings.TrimSpace(string(gitDir)), "git-http-backend"), Bindings: bindings, ObserveProvider: func(ref execution.Ref, binding nodeservices.Binding) nodeservices.Provider { + driver, e = agentruntime.New(agentruntime.Options{Root: filepath.Join(base, "nodes"), AssetsDirectory: assets, GitBackend: filepath.Join(strings.TrimSpace(string(gitDir)), "git-http-backend"), Bindings: bindings, IdleTimeout: idleTimeout, MaxIdle: idleMax, ObserveProvider: func(ref execution.Ref, binding nodeservices.Binding) nodeservices.Provider { return requestmetrics.Wrap(db, ref, binding) }, ResolveProject: func(c context.Context, scope execution.Scope, id string) (agentruntime.Project, error) { p, e := db.GetProject(c, scope, id) @@ -195,7 +199,7 @@ func run(addr, dbURL string, cleanup, fixture, testFixtures bool) (resultErr err return checkFixture(ctx, runtime, scope, demo.ID, filepath.Join(base, "nodes")) } if driver != nil { - go driver.WarmDemo(ctx, demo.ID, agentruntime.Project{MirrorPath: demo.MirrorPath, Commit: demo.Commit, SnapshotRef: demo.SnapshotRef}) + go prewarmAtStartup(ctx, db, driver, demo, idleMax) } marketURL, e := pluginmarket.NormalizeSourceURL(envOr( "V2_PLUGIN_MARKETPLACE_URL", @@ -267,6 +271,9 @@ func run(addr, dbURL string, cleanup, fixture, testFixtures bool) (resultErr err TaskRuns: taskRuns, TaskDecompositions: decompositions, }, workflowRuns) + if driver != nil { + options.ProjectCreated = warmCreated(ctx, driver) + } coordinator, e := server.NewScheduleCoordinator(db, options, time.Now) if e != nil { return e diff --git a/v2-server/prewarm.go b/v2-server/prewarm.go new file mode 100644 index 000000000..8b8d08542 --- /dev/null +++ b/v2-server/prewarm.go @@ -0,0 +1,84 @@ +//go:build linux + +package main + +import ( + "context" + "fmt" + "log/slog" + "strconv" + "time" + + "github.com/wanglongan587/cloud/v2-server/internal/agentruntime" + "github.com/wanglongan587/cloud/v2-server/internal/store" +) + +// nodeIdlePolicy reads how long an idle project Node stays up and how many idle Nodes may stay. +func nodeIdlePolicy(getenv func(string) string) (time.Duration, int, error) { + timeout, max := agentruntime.DefaultIdleTimeout, agentruntime.DefaultMaxIdle + if v := getenv("V2_NODE_IDLE_TIMEOUT"); v != "" { + d, e := time.ParseDuration(v) + if e != nil || d <= 0 { + return 0, 0, fmt.Errorf("V2_NODE_IDLE_TIMEOUT must be a positive duration") + } + timeout = d + } + if v := getenv("V2_NODE_IDLE_MAX"); v != "" { + n, e := strconv.Atoi(v) + if e != nil || n <= 0 { + return 0, 0, fmt.Errorf("V2_NODE_IDLE_MAX must be a positive integer") + } + max = n + } + return timeout, max, nil +} + +func nodeProject(p store.Project) agentruntime.Project { + return agentruntime.Project{MirrorPath: p.MirrorPath, Commit: p.Commit, SnapshotRef: p.SnapshotRef} +} + +// warmTimeout bounds one prewarm; Node itself gives up starting after 20 seconds. +const warmTimeout = time.Minute + +// prewarmAtStartup starts Nodes before anyone asks: the Demo first, then the most recently active +// projects up to the idle cap, one at a time so a 4-core demo host is not swamped while it also +// serves the first requests. A restart leaves no Node running, so without this every project's +// first conversation after it would start one. Failures are logged and left to cold start. +func prewarmAtStartup(ctx context.Context, db *store.Store, driver *agentruntime.Driver, demo store.Project, limit int) { + driver.WarmDemo(ctx, demo.ID, nodeProject(demo)) + recent, e := db.RecentProjects(ctx, limit+1) + if e != nil { + slog.Warn("list projects to prewarm", "error", e) + return + } + warmed := 0 + for _, p := range recent { + if p.ID == demo.ID || warmed == limit { + continue + } + warmed++ + warmProject(ctx, driver, p) + if ctx.Err() != nil { + return + } + } +} + +func warmProject(ctx context.Context, driver *agentruntime.Driver, p store.Project) { + ctx, cancel := context.WithTimeout(ctx, warmTimeout) + defer cancel() + started := time.Now() + if e := driver.Warm(ctx, p.ID, nodeProject(p)); e != nil { + if ctx.Err() == nil { + slog.Warn("prewarm project node", "project", p.ID, "error", e) + } + return + } + slog.Info("prewarmed project node", "project", p.ID, "duration", time.Since(started).Round(time.Millisecond)) +} + +// warmCreated prewarms a newly created project's Node in the background; the creation request +// has already succeeded and does not wait. +func warmCreated(ctx context.Context, driver *agentruntime.Driver) func(store.Project) { + return func(p store.Project) { go warmProject(ctx, driver, p) } +} diff --git a/v2-server/prewarm_test.go b/v2-server/prewarm_test.go new file mode 100644 index 000000000..b12197d62 --- /dev/null +++ b/v2-server/prewarm_test.go @@ -0,0 +1,29 @@ +//go:build linux + +package main + +import ( + "testing" + "time" + + "github.com/wanglongan587/cloud/v2-server/internal/agentruntime" +) + +func TestNodeIdlePolicy(t *testing.T) { + env := func(values map[string]string) func(string) string { + return func(key string) string { return values[key] } + } + timeout, max, err := nodeIdlePolicy(env(nil)) + if err != nil || timeout != agentruntime.DefaultIdleTimeout || max != agentruntime.DefaultMaxIdle { + t.Fatalf("defaults = %v %d %v", timeout, max, err) + } + timeout, max, err = nodeIdlePolicy(env(map[string]string{"V2_NODE_IDLE_TIMEOUT": "90m", "V2_NODE_IDLE_MAX": "3"})) + if err != nil || timeout != 90*time.Minute || max != 3 { + t.Fatalf("configured = %v %d %v", timeout, max, err) + } + for _, bad := range []map[string]string{{"V2_NODE_IDLE_TIMEOUT": "0s"}, {"V2_NODE_IDLE_TIMEOUT": "soon"}, {"V2_NODE_IDLE_MAX": "0"}, {"V2_NODE_IDLE_MAX": "-1"}} { + if _, _, err := nodeIdlePolicy(env(bad)); err == nil { + t.Fatalf("accepted %v", bad) + } + } +}