From 3d2c3987ff25650bfb441e1689a292b8f5ae6cd2 Mon Sep 17 00:00:00 2001 From: sam Date: Sun, 11 Oct 2026 20:15:51 +0800 Subject: [PATCH] fix: wake pending inputs after environment initialization --- docs/getting-started/operations.md | 2 + docs/zh/getting-started/operations.md | 4 +- services/core/IMPLEMENTATION.md | 2 + .../execution/environment_admission.go | 9 +- .../execution/execution_observation_test.go | 24 +++ .../execution/initialization_wakeup_test.go | 185 ++++++++++++++++++ .../internal/execution/prepared_dispatch.go | 7 +- .../execution/runtime_initialization.go | 6 + .../internal/execution/runtime_lifecycle.go | 10 +- 9 files changed, 245 insertions(+), 4 deletions(-) create mode 100644 services/core/internal/execution/initialization_wakeup_test.go diff --git a/docs/getting-started/operations.md b/docs/getting-started/operations.md index 23c7a2b51..41475093d 100644 --- a/docs/getting-started/operations.md +++ b/docs/getting-started/operations.md @@ -242,6 +242,8 @@ Use the `environment input reserved` log to connect the submitting HTTP trace to `execution stage` records `stage`, `duration_ms` and `status` (`ok`, `error`, `cancelled` or `timeout`). Stages cover input validation/ownership, reservation and admission wait, execution configuration and admission promotion, lifecycle gate wait, Runtime observation/provisioning, provider create/resume and Environment initialization. Runtime connection logs record confirmed observed connection transitions, not the exact socket connection instant. Use Environment, allocation and node IDs to join background Runtime records to an input; a background record without the input trace is not an unbroken request span. `runtime transport registered` records authenticated WebSocket registration. `runtime capability snapshot observed` records validated declaration changes with device ID and kind counts; transport registration alone does not establish execution readiness. Capability changes with an available Harness send a coalesced scheduling hint, and the Worker rechecks its ordinary ownership, capacity, lifecycle and capability requirements; polling recovers missed hints. Stage durations use a local monotonic clock. `queue_age_ms` instead compares selection time to the database reservation creation time and may reflect clock skew; it is not a measured scheduler-only interval. +`execution_prepare_send` measures the Core control send; `execution_prepare_wait` measures the subsequent wait for Runtime executor readiness or a terminal reservation result. Both carry Session, reservation and preparation request IDs. `environment_initialize_commit` measures the completion transaction; a successful commit wakes the scheduler, and the existing hinted scan bypasses the one-second candidate scan delay without bypassing admission checks. `input_admission_wait` includes `notification_wakeups`, `poll_wakeups` and `last_wake_source` (`none`, `notification`, `poll`); these count the branches selected by the waiting HTTP handler, not notifications sent or lost. A poll win does not prove a missing notification. `runtime_reconcile`, emitted for nonempty allocation pages, measures the observation batch and subsequent pending provisioning after inventory lookup and gate acquisition, with `allocation_count` and `observed_count`. It overlaps individual provider/observation stages and excludes earlier gate waiting. These records do not change lifecycle ordering or isolate slow allocation observations. + `control_ready_ms` starts after scheduling, Runtime readiness and configuration assembly. `start_control_ms` describes the control acknowledgement. `input_to_first_text_ms` starts immediately before execution delivery, includes start-control time and is not model-only TTFT. The admission, control and first-text intervals overlap; do not sum them as independent durations. A successful provider call describes that operation, not completed daemon connection or a completed Turn. Model request start, response headers, first model text and retries remain unavailable unless the selected native adapter provides those observations. Logs contain correlation IDs and finite status categories, not input content, credentials or raw provider errors. Daemon startup logs record `runtime process starting` with its build version and `runtime startup stage` for each Harness discovery, bootstrap and transport dial attempt. Durations use a local monotonic clock and omit credentials and raw errors. Discovery runs before transport registration. Compare daemon timestamps with Core registration and capability observations to separate startup from periodic observation and scheduling. These daemon records require a Runtime built with this instrumentation; updating Core alone does not update an existing Runtime template. A managed startup receipt confirms process spawn, not connection or capability readiness. diff --git a/docs/zh/getting-started/operations.md b/docs/zh/getting-started/operations.md index b2d7f7909..25dcf91fa 100644 --- a/docs/zh/getting-started/operations.md +++ b/docs/zh/getting-started/operations.md @@ -1,7 +1,7 @@ --- title: "运维" source: docs/getting-started/operations.md -source_hash: 54ac05e0785b1a485f93edd2b2d6bbc6bf84e36a866a442d53f4e6516813ba08 +source_hash: 493098ee6b2ebee2c6cbb1076113e381d5683625c377a338d584b51c3ebf92e9 --- 安装运维人员负责 Core 主机、存储和可用性。节点主机运行各自的服务；参阅[节点](nodes.md)。设置见[配置参考](../configuration.md)。 @@ -243,6 +243,8 @@ Web 使用 Core 密钥让管理员登录，检查每个请求来源，并用保 `execution stage` 记录 `stage`、`duration_ms` 与 `status`（`ok`、`error`、`cancelled`、`timeout`）。阶段覆盖输入验证/ownership、预留和等待 admission、执行配置与 admission 提升、生命周期 gate 等待、Runtime 观察和供给、provider create/resume、Environment 初始化。Runtime 连接日志记录已确认观察到的连接变化，不是 socket 建立的精确时间。通过 Environment、allocation、node ID 将后台 Runtime 记录与输入关联；缺少输入 Trace 的后台记录不能视为连续的请求 span。阶段耗时使用本进程单调时钟。`queue_age_ms` 则比较选择时间与数据库预留创建时间，可能受到时钟偏差影响，不能当作单独测量的调度等待。 +`execution_prepare_send` 测量 Core 控制消息发送；`execution_prepare_wait` 测量随后等待 Runtime 执行器就绪或预留进入终态的区间，两者都携带 Session、预留与准备请求 ID。`environment_initialize_commit` 测量初始化完成事务；成功提交后唤醒调度器，已有提示扫描绕过一秒候选扫描间隔，但不绕过准入检查。`input_admission_wait` 增加 `notification_wakeups`、`poll_wakeups` 和 `last_wake_source`（`none`、`notification`、`poll`）；它们统计等待 HTTP handler 实际选中的分支，不代表通知发送或丢失数量。选中轮询不证明漏通知。`runtime_reconcile` 仅对非空 allocation 页记录，测量清单查询及 gate 获取之后的观察批次和随后 pending 供给，附 `allocation_count` 与 `observed_count`；它与单独 provider/观察阶段重叠，不含之前的 gate 等待。这些记录不改变生命周期顺序，也不隔离慢实例观察。 + `control_ready_ms` 从调度、Runtime 就绪和配置组装完成后开始。`start_control_ms` 描述控制确认。`input_to_first_text_ms` 从执行投递前开始，包含 start-control 时间，不是纯模型 TTFT。admission、control、首字区间存在重叠，不要当作独立耗时相加。provider 调用成功表示该操作成功，不等于 daemon 连接完成或 Turn 完成。除非原生适配器提供相应观测，模型请求开始、响应 headers、模型首字、重试仍为不可观测。日志仅记录关联 ID 与有限状态分类，不记录输入内容、凭据或 provider 原始错误。 `runtime transport registered` 记录认证后的 WebSocket 注册，`runtime capability snapshot observed` 记录合法能力声明变化及设备 ID、能力数量。连接注册本身不代表执行就绪。包含可用 Harness 的能力变化会发出合并后的调度提示；Worker 仍检查所有权、容量、生命周期和能力要求，轮询负责兜底。 diff --git a/services/core/IMPLEMENTATION.md b/services/core/IMPLEMENTATION.md index f60275a1c..7b89096eb 100644 --- a/services/core/IMPLEMENTATION.md +++ b/services/core/IMPLEMENTATION.md @@ -78,6 +78,8 @@ Capture bytes into private large objects without a Session admission lock. Befor The leased Worker's common preparation scheduler owns Environment initialization, independently of any allocation; managed and enrolled connections use the same frozen snapshots and typed Runtime operations. Preparation has bounded concurrency separate from Turn scheduling, and a blocked Runtime never blocks cleanup or other connections. Check Harness availability before installing. A persisted running initialization whose owner is lost fails without replaying side effects; completion requires the same authorized Environment and device binding, and failure settles pending input while keeping compute ownership and user files. A connection observation never implies completion. +Successful Environment initialization sends the same scheduler hint only after its completion transaction commits, so an earlier capability hint consumed during preparation cannot leave ready input waiting for the next candidate scan. Failed completion does not announce readiness. + A committed input sends a coalesced hint to the Worker scheduler, which keeps its lease, capacity, cursor fairness and per-Session ownership checks; a hint admits nothing by itself. If capacity is occupied, keep one rescan for completion without turning a failed preparation into a busy retry loop. HTTP readiness waits and active input delivery subscribe before reading state, wake after committed promotion or input and recheck storage after each hint. Polling remains the fallback for external writers, expiry and lost hints; notifications carry no execution authority and no input data. Core readiness, Start acknowledgement and input-to-first-text logs use one process monotonic clock; a Start acknowledgement confirms adapter ownership, not model input consumption. Runtime logs report Executor creation, reuse, idle and close separately from Turn completion. Never call these durations model latency or subtract clocks of different machines. diff --git a/services/core/internal/execution/environment_admission.go b/services/core/internal/execution/environment_admission.go index c1ee9c902..6f53d19a7 100644 --- a/services/core/internal/execution/environment_admission.go +++ b/services/core/internal/execution/environment_admission.go @@ -111,9 +111,12 @@ func (w *Worker) submitEnvironmentInputs(ctx context.Context, session sessions.S obslog.Ctx(ctx).Info("environment input reserved", "session_id", session.ID, "reservation_id", reservation.ID, "execution_trace_id", reservationTraceID(reservation.ID).String(), "state", reservation.State) admittedAt := time.Now() + var notificationWakeups, pollWakeups int + lastWake := "none" defer func() { observeExecutionStage(ctx, "input_admission_wait", admittedAt, err, - "session_id", session.ID, "reservation_id", reservation.ID, "state", reservation.State) + "session_id", session.ID, "reservation_id", reservation.ID, "state", reservation.State, + "notification_wakeups", notificationWakeups, "poll_wakeups", pollWakeups, "last_wake_source", lastWake) }() w.wakeScheduler() if reservation.State == sessions.EnvironmentInputPending { @@ -136,7 +139,11 @@ func (w *Worker) submitEnvironmentInputs(ctx context.Context, session sessions.S case <-ctx.Done(): return nil, ctx.Err() case <-ticker.C: + pollWakeups++ + lastWake = "poll" case <-changed: + notificationWakeups++ + lastWake = "notification" } if err := w.checkAdmissionOwnership(ctx); err != nil { return nil, err diff --git a/services/core/internal/execution/execution_observation_test.go b/services/core/internal/execution/execution_observation_test.go index 45eae02e3..b4e1bbba1 100644 --- a/services/core/internal/execution/execution_observation_test.go +++ b/services/core/internal/execution/execution_observation_test.go @@ -176,3 +176,27 @@ func TestInitialInputOriginKeepsHTTPTraceAndSessionIdentity(t *testing.T) { t.Fatal("creation fabricated a reservation identity") } } + +func TestAdmissionStageDiagnosticsUseFiniteStatusAndSafeFields(t *testing.T) { + out := captureExecutionLogs(t) + stages := []string{"execution_prepare_send", "execution_prepare_wait", "environment_initialize_commit", "runtime_reconcile"} + for _, stage := range stages { + for _, sample := range []struct { + err error + status string + }{{nil, "ok"}, {context.Canceled, "cancelled"}, {context.DeadlineExceeded, "timeout"}, {errors.New("SECRET fixture error body"), "error"}} { + out.Reset() + observeExecutionStage(reservationTrace(t.Context(), "reservation"), stage, time.Now(), sample.err, "session_id", "session", "reservation_id", "reservation") + var entry map[string]any + if err := json.Unmarshal(out.Bytes(), &entry); err != nil { + t.Fatal(err) + } + if entry["stage"] != stage || entry["status"] != sample.status || entry["duration_ms"].(float64) < 0 || entry["reservation_id"] != "reservation" { + t.Fatal("invalid stage schema", entry) + } + if strings.Contains(out.String(), "SECRET") || strings.Contains(out.String(), "fixture error body") { + t.Fatal("raw failure leaked") + } + } + } +} diff --git a/services/core/internal/execution/initialization_wakeup_test.go b/services/core/internal/execution/initialization_wakeup_test.go new file mode 100644 index 000000000..372425ff0 --- /dev/null +++ b/services/core/internal/execution/initialization_wakeup_test.go @@ -0,0 +1,185 @@ +package execution + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto/prototest" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/environmentconfig" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimedevice" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" + "github.com/gorilla/websocket" +) + +type initializationWakeReader struct { + sessions.Reader + committed bool + scans int +} + +func (*initializationWakeReader) GetEnvironment(context.Context, string, string) (sessions.Environment, error) { + return sessions.Environment{ID: "environment", Configuration: json.RawMessage(`{"type":"self_hosted","workspace_directory":"/workspace"}`)}, nil +} +func (*initializationWakeReader) ReadEnvironmentSetup(context.Context, string, string) (environmentconfig.Setup, error) { + return environmentconfig.Setup{}, nil +} +func (*initializationWakeReader) GetDeviceCredential(context.Context, string) (runtimedevice.Credential, bool, error) { + return runtimedevice.Credential{ID: "device", Type: runtimedevice.RuntimeTypeAgentDaemon, CredentialHash: runtimedevice.HashCredential("fixture")}, true, nil +} +func (*initializationWakeReader) ListExecutionWork(context.Context, string, []string, []string) ([]sessions.ExecutionWork, error) { + return nil, nil +} +func (r *initializationWakeReader) ListEnvironmentInputWork(context.Context, string, []string) ([]sessions.EnvironmentInputWork, error) { + r.scans++ + if !r.committed { + return nil, nil + } + return []sessions.EnvironmentInputWork{{TenantID: "tenant", SessionID: "session", ReservationID: "reservation"}}, nil +} + +func (r *initializationWakeReader) GetSession(context.Context, string, string) (sessions.Session, error) { + return sessions.Session{ID: "session", TenantID: "tenant", Engine: "codex", Configuration: json.RawMessage(`{"agent":{"model":"fixture"},"environment":{"type":"self_hosted","workspace_directory":"/workspace"}}`)}, nil +} +func (r *initializationWakeReader) GetSessionEnvironment(ctx context.Context, _, _ string) (sessions.Environment, error) { + e, _ := r.GetEnvironment(ctx, "", "") + e.SessionID = "session" + e.TenantID = "tenant" + return e, nil +} +func (*initializationWakeReader) GetSessionDevice(context.Context, string, string) (sessions.ExecutionDevice, error) { + return sessions.ExecutionDevice{ID: "device", EnvironmentID: "environment"}, nil +} + +type initializationWakeTx struct{ sessions.InitializationTx } + +func (*initializationWakeTx) LoadEnvironment(context.Context) (sessions.Environment, error) { + return sessions.Environment{ID: "environment"}, nil +} +func (*initializationWakeTx) LoadSessionDevice(context.Context) (sessions.ExecutionDevice, bool, error) { + return sessions.ExecutionDevice{ID: "device", EnvironmentID: "environment"}, true, nil +} +func (*initializationWakeTx) CompleteInitialization(context.Context, string) (bool, error) { + return true, nil +} + +type initializationWakeStorage struct { + sessions.ExecutionStorage + reader *initializationWakeReader + worker *Worker + calls int + commitError error +} + +func (s *initializationWakeStorage) WithInitialization(ctx context.Context, _, _ string, apply func(context.Context, sessions.InitializationTx, sessions.LockedSession) error) error { + s.calls++ + if s.calls > 1 { + return s.commitError + } // The failure recorder must not fabricate readiness. + if err := apply(ctx, &initializationWakeTx{}, sessions.LockedSession{}); err != nil { + return err + } + if len(s.worker.scheduleWake) != 0 { + return errors.New("scheduler woke before initialization commit") + } + if s.commitError != nil { + return s.commitError + } + s.reader.committed = true + return nil +} + +// The initial capability hint can be consumed while initialization is pending. +// A committed initialization must trigger another actual candidate scan without +// waiting for the one-second Environment scan deadline; a failed commit must not. +func TestInitializationCommitWakesThrottledScheduler(t *testing.T) { + for _, failed := range []bool{false, true} { + t.Run(map[bool]string{false: "committed", true: "commit_failed"}[failed], func(t *testing.T) { + reader := &initializationWakeReader{} + registry := runtimegateway.NewRegistry() + handler := runtimegateway.NewHandler(runtimegateway.HandlerConfig{Authenticator: runtimegateway.NewAuthenticator(reader), Registry: registry}) + server := httptest.NewServer(http.HandlerFunc(handler.WS)) + defer server.Close() + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http")+"?device_id=device&version="+proto.Version, http.Header{"Authorization": {"Bearer fixture"}}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err = registry.WaitForDevice(t.Context(), "device", time.Second); err != nil { + t.Fatal(err) + } + heartbeat, _ := proto.NewEnvelope(proto.TypeHeartbeat, "", proto.HeartbeatPayload{SupportedAgentKinds: []proto.SupportedAgentKind{{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{Streaming: proto.CapabilitySupported, Steering: proto.CapabilitySupported, DurableTurns: proto.CapabilitySupported, DurableInputReceipts: proto.CapabilitySupported, Preparation: proto.CapabilitySupported, ExecutionControls: proto.CapabilitySupported, WebSearchControl: proto.CapabilitySupported, TextVerbosity: proto.CapabilitySupported, ToolObservations: proto.CapabilitySupported, SubagentControl: proto.CapabilitySupported, LocalEnvironment: proto.CapabilitySupported, WorkspaceReadPreparation: proto.CapabilitySupported, WorkspaceOutputExport: proto.CapabilitySupported})}}}) + if err = conn.WriteJSON(heartbeat); err != nil { + t.Fatal(err) + } + select { + case <-registry.CapabilityHints(): + case <-time.After(time.Second): + t.Fatal("capability hint absent") + } + worker := &Worker{dispatcher: &Dispatcher{SessionsReader: reader, Registry: registry, DeploymentReader: &strictDeploymentReader{t: t, environmentAllocation: func(context.Context, deployment.AllocationKey) (deployment.Allocation, error) { + return deployment.Allocation{}, deployment.ErrNotFound + }}}, scheduleWake: make(chan struct{}, 1), concurrency: 1} + storage := &initializationWakeStorage{reader: reader, worker: worker} + if failed { + storage.commitError = errors.New("fixture commit failure") + } + worker.dispatcher.sessionExecution, err = sessions.NewExecutionOperations(storage) + if err != nil { + t.Fatal(err) + } + schedule := workerSchedule{} + if work, err := schedule.selectWork(t.Context(), worker, []string{"device"}, map[string]bool{}, true); err != nil || len(work) != 0 { + t.Fatal("pending initialization selected work", err) + } + reader.scans = 0 + schedule.nextEnvironmentScan = time.Now().Add(time.Hour) + if _, err = schedule.selectWork(t.Context(), worker, []string{"device"}, map[string]bool{}, false); err != nil { + t.Fatal(err) + } + if reader.scans != 0 { + t.Fatal("baseline poll bypassed scan deadline") + } + worker.initializeEnvironment(t.Context(), sessions.EnvironmentInitialization{EnvironmentID: "environment", SessionID: "session", DeviceID: "device"}) + if failed { + if len(worker.scheduleWake) != 0 || reader.committed { + t.Fatal("failed commit announced readiness") + } + return + } + if !reader.committed { + t.Fatal("initialization did not commit") + } + select { + case <-worker.scheduleWake: + default: + t.Fatal("committed initialization waits for fallback instead of waking scheduler") + } + active := map[string]bool{} + work, err := schedule.selectWork(t.Context(), worker, []string{"device"}, active, true) + if err != nil || len(work) != 1 || work[0].reservationID != "reservation" { + t.Fatal("committed input not selected", work, err) + } + if duplicate, err := schedule.selectWork(t.Context(), worker, []string{"device"}, active, true); err != nil || len(duplicate) != 0 { + t.Fatal("duplicate dispatch", duplicate, err) + } + if reader.scans != 2 { + t.Fatal("ready hint did not bypass Environment scan throttle") + } + if _, err = schedule.selectWork(t.Context(), worker, []string{"device"}, map[string]bool{}, false); err != nil { + t.Fatal(err) + } + if reader.scans != 2 { + t.Fatal("ready hint disabled polling throttle") + } + }) + } +} diff --git a/services/core/internal/execution/prepared_dispatch.go b/services/core/internal/execution/prepared_dispatch.go index 69d901253..8e9d9a94b 100644 --- a/services/core/internal/execution/prepared_dispatch.go +++ b/services/core/internal/execution/prepared_dispatch.go @@ -94,10 +94,15 @@ func (d *Dispatcher) RunEnvironmentInput(ctx context.Context, lease Ownership, t defer prepared.close() obslog.Ctx(owner).Info("execution preparation requested", "session_id", sessionID, "reservation_id", reservationID, "environment_id", environment.ID, "device_id", bound.Device.ID, "preparation_request_id", prepared.requestID) - if err = send(owner, peer, proto.TypeExecutionPrepare, prepared.requestID, proto.ExecutionPreparePayload{SessionID: sessionID, Configuration: req}); err != nil { + prepareAt := time.Now() + err = send(owner, peer, proto.TypeExecutionPrepare, prepared.requestID, proto.ExecutionPreparePayload{SessionID: sessionID, Configuration: req}) + observeExecutionStage(owner, "execution_prepare_send", prepareAt, err, "session_id", sessionID, "reservation_id", reservationID, "preparation_request_id", prepared.requestID) + if err != nil { return run, err } + prepareAt = time.Now() run.Reservation, err = d.awaitPreparation(owner, tenantID, sessionID, run.Reservation, prepared) + observeExecutionStage(owner, "execution_prepare_wait", prepareAt, err, "session_id", sessionID, "reservation_id", reservationID, "preparation_request_id", prepared.requestID, "state", run.Reservation.State) if err != nil || run.Reservation.State != sessions.EnvironmentInputPending { return run, err } diff --git a/services/core/internal/execution/runtime_initialization.go b/services/core/internal/execution/runtime_initialization.go index 880175455..cf5745860 100644 --- a/services/core/internal/execution/runtime_initialization.go +++ b/services/core/internal/execution/runtime_initialization.go @@ -98,7 +98,13 @@ func (w *Worker) initializeEnvironment(ctx context.Context, owner sessions.Envir initializeAt := time.Now() err := w.prepareEnvironment(operation, owner, &failure) if err == nil { + commitAt := time.Now() err = w.dispatcher.sessionExecution.CompleteEnvironmentInitialization(operation, owner) + observeExecutionStage(ctx, "environment_initialize_commit", commitAt, err, "environment_id", owner.EnvironmentID, "session_id", owner.SessionID, "device_id", owner.DeviceID) + if err == nil { + // Capabilities may have arrived before initialization committed. + w.wakeScheduler() + } } observeExecutionStage(ctx, "environment_initialize", initializeAt, err, "environment_id", owner.EnvironmentID, "session_id", owner.SessionID, "device_id", owner.DeviceID) if err != nil { diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index 0916e70dc..8b0af5a08 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -322,7 +322,7 @@ func (w *Worker) ReconcileManagedRuntimes(ctx context.Context) error { return w.runtimes.reconcile(ctx) } -func (r *runtimeLifecycle) reconcile(ctx context.Context) error { +func (r *runtimeLifecycle) reconcile(ctx context.Context) (scanErr error) { ctx, finish, err := r.beginReconcile(ctx) if err != nil { return err @@ -344,7 +344,15 @@ func (r *runtimeLifecycle) reconcile(ctx context.Context) error { } } } + scanAt := time.Now() + observed := 0 + defer func() { + if len(rows) > 0 { + observeExecutionStage(ctx, "runtime_reconcile", scanAt, scanErr, "node_id", r.nodeID, "allocation_count", len(rows), "observed_count", observed) + } + }() for _, owner := range rows { + observed++ r.cursor = owner.ID operation, stop := context.WithTimeout(ctx, 30*time.Second) observeAt := time.Now()