From e60daaf2c6fc7e6b412b8f669d353054d65cba4e Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Tue, 22 Sep 2026 21:27:40 +0800 Subject: [PATCH] fix: avoid empty managed runtime scan intervals --- CONTRIBUTING.md | 6 +- .../internal/execution/runtime_lifecycle.go | 10 +- .../store/runtime_initialization_test.go | 15 ++- .../internal/store/runtime_scan_test.go | 123 ++++++++++++++++++ 4 files changed, 147 insertions(+), 7 deletions(-) create mode 100644 services/agents-api/internal/store/runtime_scan_test.go diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index cb69849b7..fa4296efa 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -303,8 +303,10 @@ The allocation lifecycle owns pending/running/complete initialization. Authentic may connect the daemon during initialization; execution bindings, native preparation, live Files and connected publication wait for completion. Keep Provider bootstrap settlement distinct. Advance at most one bounded initialization operation per full -maintenance scan, -using process-local progress and the existing lifecycle gate. A recovered or uncertain +maintenance scan. At allocation EOF, begin the next page in the same call rather +than consume an observation interval on an empty page. Refill at most once, retain +the 32-allocation per-call bound and the five-second ticker, and never loop on an +empty store. Use process-local progress and the existing lifecycle gate. A recovered or uncertain running installation fails and uses existing cleanup, without replaying writes. Completed environments never reinstall initial files on reconnect or native recovery. Provider RunCommand carries bounded stdin, not confidential argv. Only fixed trusted diff --git a/services/agents-api/internal/execution/runtime_lifecycle.go b/services/agents-api/internal/execution/runtime_lifecycle.go index 559910fbb..6b81261aa 100644 --- a/services/agents-api/internal/execution/runtime_lifecycle.go +++ b/services/agents-api/internal/execution/runtime_lifecycle.go @@ -188,6 +188,7 @@ func (w *Worker) ReconcileManagedRuntimes(ctx context.Context) error { return err } if len(rows) == 0 { + wrapped := r.cursor != "" r.cursor = "" if err := r.advanceInitialization(ctx); err != nil { if ownership := r.store.CheckExecutionOwnership(ctx); ownership != nil { @@ -195,7 +196,14 @@ func (w *Worker) ReconcileManagedRuntimes(ctx context.Context) error { } log.Ctx(ctx).Warn("managed Runtime file initialization incomplete") } - return r.provisionPending(ctx) + if wrapped { + // Service the next page now instead of spending a ticker interval on EOF. + // Refill only once so an empty store still returns without spinning. + rows, err = r.store.ListRuntimeAllocations(ctx, "") + if err != nil { + return err + } + } } for _, owner := range rows { r.cursor = owner.ID diff --git a/services/agents-api/internal/store/runtime_initialization_test.go b/services/agents-api/internal/store/runtime_initialization_test.go index c01da1502..445dc2b04 100644 --- a/services/agents-api/internal/store/runtime_initialization_test.go +++ b/services/agents-api/internal/store/runtime_initialization_test.go @@ -92,7 +92,12 @@ func TestManagedInitialFilesGateFairnessCompletionAndRestart(t *testing.T) { if err != nil || !ok || credential.ID != owner.DeviceID { t.Fatal("pending initialization blocks daemon authentication") } + lastStepGets := 0 p.check = func() { + if p.gets-lastStepGets < 33 { + t.Fatal("initialization advanced before a full allocation scan", p.gets-lastStepGets) + } + lastStepGets = p.gets if _, err := s.GetSessionDevice(t.Context(), tenant, session.ID); !errors.Is(err, store.ErrNotFound) { t.Fatal("pending file access", err) } @@ -100,10 +105,12 @@ func TestManagedInitialFilesGateFairnessCompletionAndRestart(t *testing.T) { t.Fatal("premature native preparation", err) } } - // Another allocation is observed between file steps, rather than after the full batch. - otherTenant, _, otherEnv := managedSession(t, s) - if _, err := w.ProvisionEnvironment(t.Context(), otherTenant, otherEnv.ID, key); err != nil { - t.Fatal(err) + // A full page of other allocations is serviced between initialization steps. + for range 32 { + otherTenant, _, otherEnv := managedSession(t, s) + if _, err := w.ProvisionEnvironment(t.Context(), otherTenant, otherEnv.ID, key); err != nil { + t.Fatal(err) + } } for n := 0; p.writes == 0 && n < 100; n++ { if err := w.ReconcileManagedRuntimes(t.Context()); err != nil { diff --git a/services/agents-api/internal/store/runtime_scan_test.go b/services/agents-api/internal/store/runtime_scan_test.go new file mode 100644 index 000000000..c868670d8 --- /dev/null +++ b/services/agents-api/internal/store/runtime_scan_test.go @@ -0,0 +1,123 @@ +package store_test + +import ( + "context" + "errors" + "fmt" + "slices" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/sandbox" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" +) + +type scanProvider struct { + lifecycleProvider + observed []string +} + +func (p *scanProvider) GetInfo(ctx context.Context, ref sandbox.Reference) (sandbox.Info, error) { + p.observed = append(p.observed, ref.AllocationID) + return p.lifecycleProvider.GetInfo(ctx, ref) +} + +func TestManagedRuntimeScanWrapServicesNextPage(t *testing.T) { + for _, count := range []int{0, 1, 31, 32, 33, 65} { + t.Run(fmt.Sprint(count), func(t *testing.T) { + s, _ := store.NewManagedTestStore(t) + p := &scanProvider{lifecycleProvider: lifecycleProvider{resources: map[string]sandbox.Info{}}} + key := uuid.NewString() + w, _ := managedWorker(t, s, key, p) + var ids []string + for range count { + tenant, _, env := managedSession(t, s) + owner, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key) + if err != nil { + t.Fatal(err) + } + ids = append(ids, owner.ID) + } + slices.Sort(ids) + // Each call must service the next bounded page, including the call + // that wraps at EOF. Empty stores must return without spinning. + pages := max(1, (count+31)/32) + for call := 0; call < pages*3; call++ { + p.observed = nil + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + err := w.ReconcileManagedRuntimes(ctx) + cancel() + if err != nil { + t.Fatal(err) + } + start := (call % pages) * 32 + want := ids[start:min(start+32, count)] + if !slices.Equal(p.observed, want) { + t.Fatalf("call %d observed %v, want %v", call, p.observed, want) + } + } + if p.creates != count || p.kills != 0 { + t.Fatal("scanning changed resource ownership", p.creates, p.kills) + } + }) + } +} + +func TestManagedRuntimeScanEmptyAfterCleanupAndCanceledCall(t *testing.T) { + s, _ := store.NewManagedTestStore(t) + p := &scanProvider{lifecycleProvider: lifecycleProvider{resources: map[string]sandbox.Info{}}} + key := uuid.NewString() + w, _ := managedWorker(t, s, key, p) + tenant, session, env := managedSession(t, s) + owner, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key) + if err != nil { + t.Fatal(err) + } + if err := w.ReconcileManagedRuntimes(t.Context()); err != nil { + t.Fatal(err) + } + p.observed = nil + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := w.ReconcileManagedRuntimes(ctx); !errors.Is(err, context.Canceled) { + t.Fatal("canceled scan", err) + } + if len(p.observed) != 0 { + t.Fatal("canceled scan reached provider") + } + if err := s.DeleteSession(t.Context(), tenant, session.ID); err != nil { + t.Fatal(err) + } + // EOF must not delay cleanup to a second call, and cancellation must + // release the lifecycle gate. + if err := w.ReconcileManagedRuntimes(t.Context()); err != nil { + t.Fatal(err) + } + got, err := s.GetRuntimeAllocation(t.Context(), tenant, env.ID) + if err != nil || got.State != "released" || p.kills != 1 { + t.Fatal("cleanup delayed at EOF", got, err, p.kills) + } + for range 2 { + p.observed = nil + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + err := w.ReconcileManagedRuntimes(ctx) + cancel() + if err != nil || len(p.observed) != 0 { + t.Fatal("released allocation reobserved", owner.ID, err, p.observed) + } + } + // A new allocation remains discoverable after the store becomes empty. + tenant, _, env = managedSession(t, s) + next, err := w.ProvisionEnvironment(t.Context(), tenant, env.ID, key) + if err != nil { + t.Fatal(err) + } + if err := w.ReconcileManagedRuntimes(t.Context()); err != nil { + t.Fatal(err) + } + if !slices.Equal(p.observed, []string{next.ID}) { + t.Fatal("new allocation stranded after empty scan", p.observed) + } +}