Repository navigation
Conversation
Context gains StepStat (name/took_us/error) with AddStepStat/ GetStepStats, and the sequential runner times each task so callers (document enrichment timelines in coco) can record which processors ran, for how long, and which failed. Matching the API the coco feat/knowledge-hub branch consumes; async runners record nothing yet.
…observability
Record model:
- RecordCodec registry (Namespace, error-returning, OrReplace, discovery)
- ExecutionObserver + StepEvent (one instrumentation, three projections)
- trace_id helpers (SetTraceID/TraceID on record metadata)
- pipelinelog helpers (auto prefix [tr-xxx] for log correlation)
Contract system:
- RecordContract{Scope, Namespace, RequiredFields, Granularity}
- NewPipelineWithScope: compile-time placement validation
- Granularity: record / batch_capable / batch_only
- Stage-wise execution (written order = execution order)
- Catalog exposes scope + record_contract + granularity
Lane (in-process record channel):
- core/lane: bounded channel, block/drop/spill backpressure
- EmitAck with receipts (settlement on processing completion)
- SubscribeBatch (batch settlement point)
- lane_consumer host (batch accumulate → stage-wise sub-chain → output bridge)
- Unified output processor (one-of queue/lane target, ack support)
- tap processor (sampled side-copy, main flow unaffected)
Observability:
- dryrun package (snapshots, field diffs, side-effect capture)
- POST /pipeline/dry-run (unified, single definition)
- Flight recorder (ring 1024, errors always, OK 1/100 sampled)
- StageStats aggregation + GET /pipeline/task/:id/stages
- GET /pipeline/lanes (depth/rates/drops/spills)
- GET /pipeline/codecs (discovery)
- _pause/_resume (batch-boundary park) / _inject (sample injection)
- Metrics persistence (PipelineStageStat ORM, periodic snapshot)
- pipelinelog (trace_id-prefixed logging)
Version management:
- PipelineVersion (auto-snapshot on update/delete, rollback)
- Transient pipelines skip versioning
Performance:
- for_each.parallel (record-level workers, default 1)
- processHistory O(n²) fix (32-entry cap on flow trace)
- indexing_merge partial _id extraction (jsonparser zero-alloc)
- JSON nil-value guards for jsoniter compatibility
PipelineDeployment model: - PlacementSpec: selector + strategy (singleton/all) + rollout (canary/batch) - Version increments on spec change (drift detection) Reconciler: - 15s ticker, iterates enabled deployments - v1: logs convergence intent (configs-center push is the next milestone) Fleet API: - CRUD: POST/GET/PUT/DELETE /pipeline/deployments/ - GET /pipeline/cluster/overview (deployment × instance matrix) - All endpoints with API + UI permission gating Placement translation and instance reporting (DeploymentStatus) are stubbed — the configs-center integration (D7.6) is the next step.
The reconciler now writes actual ManagedConfig rows (the same transport Logpilot stream publish uses — instances pull via _sync and hot-reload): PipelineDeployment → reconcile → ManagedConfig row → instance _sync - Upsert semantics: find existing row by name, bump version, save - Readonly=true (deployment-owned; manual edits are overwritten) - singleton strategy translates to singleton:true in the config content - v1: broadcast to all instances (selector targeting is next milestone)
Instance runtime reporting (desired-vs-reported loop): - ConfigSyncRequest extended with Runtime field (PipelineRuntimeState) - Configs server persists runtime_pipelines on the instance record - PipeModule.RuntimeSnapshot() exposes live task states for the sync client Deployment status: - GET /pipeline/deployments/:id/status → DeploymentStatus (desired count, ready count, per-instance reports) - GET /pipeline/cluster/overview enriched with live status per deployment - Reconciler reads instance RuntimePipelines to compute reported state Full P9 control loop now closed: desired (Deployment) → reconcile → managed config push → instance _sync → runtime report → status → cluster overview
PipelineVersion, PipelineDeployment, PipelineStageStat need explicit orm.MustRegisterSchemaWithIndexName to auto-create their SQLite/ES tables on module startup — without it queries fail with 'no such table'.
Without orm.MustRegisterSchemaWithIndexName, the pipelines/_search endpoint fails with 'no such table' (500).
orm.Save requires an explicit ID; without it the create handler returns 500 'id was not found'.
The sync client now includes RuntimeReport (which pipeline tasks are running) in every _sync request. The configs server persists it on the instance record; the deployment reconciler reads it to compute DeploymentStatus (health: ready/desired). Closes the P9 control loop: desired → push config → instance sync → runtime report → status
PlacementSpec.Instances lists explicit target instance IDs; the reconciler creates per-instance managed configs instead of broadcast. Empty list = broadcast to all (backward compatible). UI passes the selected target instance from the deployment dialog.
The output processor's queue field expects *queue.QueueConfig (object),
but pipelines written by hand use the short form {"output":{"queue":"name"}}.
QueueRef unpacks both the bare string and the full object.
…ents - add orm.IsHandlerRegistered() so instances without an ORM backend (agents running only pushed managed configs) can check availability - reconcileOnce now returns early instead of panicking on "ORM handler is not registered"
- reconcileDeployment skips the version bump and managed-config push when the rendered content is unchanged (was re-pushing every 15s, rewriting files on every instance each cycle) - RuntimeSnapshot reports under the pipeline NAME (the contexts map key is the config ID which only equals the name for id-less YAML configs) - getDeploymentStatus resolves the pipeline name from the config row before matching instance runtime reports
…ted or distributed
Configs referencing unknown processors or carrying invalid parameters
were accepted by the control plane and only failed on the instance
after distribution. Add ValidatePipelineConfig — a compile-without-run
pass covering DSL shape, registry existence, parameter unpacking,
nested chains and record-contract placement — and wire it as a gate:
- POST/PUT /pipelines/ reject invalid configs with structured errors
(stage index + processor key + message) before persistence
- POST /pipeline/task create validates disabled pipelines too
- POST /pipeline/deployments/ requires the referenced pipeline to
exist and compile (fail-closed on read errors)
- the reconciler refuses to distribute configs that stopped compiling
(e.g. processor plugin removed after an upgrade)
- new POST /pipeline/_validate pre-flight endpoint for editors:
200 report {valid, errors, warnings}, nothing persisted
Contract mismatches surface as warnings unless the config opts into
strict_contract, matching runtime compile semantics.
…d fields enforced The compile gate only catches what constructors check. Audited every registered processor by constructing each with an empty config: - constructor panics are config errors, not runtime faults: consumer (empty inner chain), indexing_merge / json_indexing / merge_to_bulk (missing input_queue / output_queue.name / index_name, producer acquisition) now return errors instead of panicking the host - http requires at least one host (empty hosts silently failed every record; templates like $[[env.HOSTS]] stay non-empty strings) - smtp requires servers[].server.host and port - ValidatePipelineConfig recovers per-step constructor panics and reports them as validation errors — the gate itself cannot crash - exported pipeline.ProcessorConstructors() for catalog enumeration - TestEmptyConfigAudit pins the catalog: empty-accept is legitimate only for processors whose every parameter has a working default (44 audited entries in the documented allowlist); new processors must either validate required fields or join the allowlist
- rollback restores old versions through the compile gate: a version referencing a processor that was renamed/removed since must not be restored into a broken live config - pipelines whose config fails to compile on an instance (e.g. missing processor in the local catalog) now surface as FAILED + reason in the runtime report instead of being silently absent — previously only a local log line recorded why - Reason flows instance snapshot → _sync runtime report → persisted instance record → deployment status InstanceReport, so operators see per-instance failure causes without shell access
managed configs pushed by the deployment reconciler render the queue target as a plain string; hand-written configs use the full object form — the output constructor now resolves both
Manager callbacks proxied through the reverse channel execute as loopback HTTP calls against the agent's own web port, authenticated with an AGENT_API_ACCESS_TOKEN from the keystore — but nothing ever minted that token, so every callback died with 401 invalid login. - Setup registers the declared instance-ops permission keys (they are constants never route-registered, so the permission filter's Must-resolve of system:* route options panicked with 500 instead) - execute() now validates the stored token (KV record present AND carrying every instance-ops permission) and re-mints when stale — e.g. tokens left by earlier builds or wiped data dirs - the minted token carries instance-ops + all registered permissions, never expires, and is cached in the keystore for restart reuse
…d contracts, loopback token - pipeline.md: Configuration Validation section (ValidatePipelineConfig findings, the six enforcement points, /pipeline/_validate pre-flight, the constructor contract behind TestEmptyConfigAudit); Pipeline Configuration API (CRUD + versions + validated rollback); Pipeline Deployment / Fleet (reconciler convergence, runtime report with FAILED+reason); Runtime APIs table (stages, flight, inject, dry-run, codecs, lanes, cluster overview) - pipeline_record.md: Record Contracts — RecordContract fields, ContractProvider, granularity table (record/batch_capable/batch_only), compile-time placement vs runtime RequiredFields guard - websocket_reverse.md: Security Model now documents the auto-minted AGENT_API_ACCESS_TOKEN loopback credential and the instance-ops permission registration that makes route resolution succeed
…fig recall Disabling a deployment previously stopped reconciliation only — the pushed managed config stayed live and the pipeline KEPT RUNNING on every instance. Now: - reconcileOnce processes ALL deployments (disabled included); a disabled one keeps convering with enabled:false + auto_start:false in the rendered config — the instance's config watcher stops and cleans the task while keeping the file, so re-enabling is instant - deleteDeploymentHandler cascades: when no other deployment references the same pipeline, the pushed managed config row is removed too (shared references are protected) — instances see the file Deleted on next sync and the watcher removes + stops the task; previously the row leaked with a TODO
The reconciler previously ignored placement.selector/instances and always broadcast a single '*' managed config. It now resolves the scope against the registered fleet (label AND-match + explicit IDs union, broadcast when neither is set) and converges the managed-config row set EXACTLY: one row per target instance, rows for instances that left every scope are deleted so their next sync recalls the file and stops the task. - deployments sharing a pipeline merge target sets (enabled outranks disabled, singleton wins; any broadcast subsumes targeting) - row versions stay monotonic per config name so migrated rows land above what clients previously pulled (sync applies newest-per-name) - legacy '*' rows migrate in place onto per-instance rows - PUT re-validates pipeline exists + compiles (edit can rebind it); strategy whitelist all|singleton; empty selector keys dropped - status computes desired_count from the resolved scope, reports in-scope instances as PENDING before first runtime data, and marks out-of-scope runners stale (recall in flight/failed) - delete cascade removes every row of the config name (was: '*' only) Covered by deployment_targeting_test.go: resolve modes, label-scope exact convergence without version churn, scope-change recall with version monotonicity, broadcast→targeted migration.
Chained if/then/else blocks get unwieldy once one input routes to three
or more destinations. The new `case` step is an ordered multi-way
dispatch where the FIRST matching branch runs (exactly one arm), with an
optional `default` chain:
- case:
field: log_level
branches:
- values: [error, fatal] # membership, like the `in` condition
then: [...]
- when: # any conditions.Config operator
prefix: {log_level: wa}
then: [...]
default: [...] # optional; no-match without it is a no-op
Arm chains compile through newPipelineWithScope so they inherit the
enclosing scope (record contracts, batch_only placement rules), exactly
like if/then/else arms. Value resolution mirrors if/then/else too: the
current record takes precedence, context fallback at pipeline level.
Which arm ran is recorded in the flow trace as case:<n>, case:default,
or case:no-match. Compile errors carry the offending branch index.
- lane.GetOrCreate replaces a registered-but-CLOSED lane instead of returning it: consumers close their lane on task exit, and without replacement every producer would emit-fail 'closed' forever with no way for a consumer restart to recover (double-checked under the registry write lock) - safe-read helpers with tests; vfs and instance-model extensions - otel: log-record conversions (+tests) - cluster secrets/loader hardening; host utilization stats counters
A consumer parked at the write head whose offset's data files are gone no longer errors out: re-open the segment file if it exists by now and continue, otherwise report an empty fetch so the consumer keeps waiting. openReadFile factors the shared tail of ResetOffset and the self-healing reset paths. Covered by consumer_selfheal_test.
Same-name rows resolve by specificity: per-instance rows (edge-rule variants) shadow group/catch-all rows for that instance; heartbeat keeps server-owned Alias/Description display metadata.
New memory-path bulk pipeline: merge_to_bulk_lane merges records onto a lane; bulk_indexing_lane subscribes to the merged-bulk lane and ships each payload to Elasticsearch with batch-level receipt settle (the memory-path twin of bulk_indexing). Lane consumer lifecycle adjusted for restart; conditionals inside per-record sub-chains (for_each, direct-ship) now see the current record's fields; indexing_merge aligned with the lane path.
Query builders could only place parameters inside the bool clause, so body-root settings like min_score had no home — nestling them into the clause makes Elasticsearch reject the search with "unknown field [min_score]". QueryBuilder gains BodyParameter(key, val): values recorded there are rendered by the Elasticsearch DSL builder at the body root, next to query/from/size, in BuildQueryDSL — and through the OnTopOfDSL paths, where the explicit parameter also beats a raw request-body key of the same name (plain Parameter keeps feeding the clause unchanged). Needed by coco's document search quality loop, which sets a search-body-level min_score for semantic/hybrid retrieval.
- PipelineConfigV2 ORM mappings: Name becomes a keyword field so the pipeline list API can term-query and sort on it; Processors disables indexing — processor configs are arbitrary per-processor JSON and dynamic mapping would risk field-type conflicts between pipelines sharing the index. - replay: req/res Reset moves from execute's defer to the callers — execute logs the request when it fails, and must do so while the request is still intact instead of after the deferred reset emptied it.
Firecracker-style loss accounting for the observability buffers: losing telemetry under backpressure is fine, losing track of how much was lost is not. - flight recorder: sampled_out counts OK steps skipped by the 1/N sampling, evicted counts entries pushed out of a full ring (both atomic, cumulative since process start); the flight endpoint returns them in a `lost` block next to the events. - metrics exporter: persist_failures counts stage-snapshot rows a failed orm save dropped; the stages endpoint surfaces it as persist_errors so a non-zero value marks holes in the persisted history.
…s don't panic CreateAPIToken/ListTokens/DeleteTokenByID built direct contexts without orm.WithModel; the elastic ORM's data-op hooks resolve the index from the ctx model and panicked 'model is not set'. SQLite-backed products (logpilot) resolve from the object and never saw it — minting an instance manager token on coco (native + elastic) died mid-approve.
A YAML/JSON value of {} reified to nil, and reflect.ValueOf(nil) is an
invalid Value, so reifyMap skipped SetMapIndex and the KEY disappeared
from any map[string]interface{} target. Config-file pipelines with empty
processor bodies (processor: [{echo: {}}] — the most common shape) lost
the processor name at load and failed with 'each processor must have
exactly one action, but found 0 actions'; managed instances could never
start deployed pipelines. Empty dicts now reify to empty maps, mirroring
the existing empty-array behavior ([]interface{}{}).
…beats A binary built without version ldflags reports build_date/eol_date as "N/A". On indices whose coco_instance mapping predates the keyword pin, those fields carry a dynamic date mapping and ES rejects the upsert with document_parsing_exception — every heartbeat failed silently (Debug log) while /configs/_sync kept returning 200, so the instance showed offline forever. Three layers: - env.Version fields now pin keyword elastic_mapping (new indices) - upsertInstance blanks non-RFC3339 date placeholders before Save (protects existing date-mapped indices) - heartbeat upsert failure logs at Warn instead of Debug
Drops the sampled-out/evicted flight counters and FlightLostStats, the metrics persist-failure counter, and the lost/persist_errors fields the flight and stages endpoints carried. The 1/N sampling loss is by design; the loss counters did not earn their complexity.
Two silent gaps made the strict-contract machinery a no-op for real product pipelines: - validatePlacement only consulted the constructed instance's ContractProvider method, but every product processor declares its contract through RegisterProcessorContract — registry declarations never drove placement. The registry is now the fallback source, so record-scoped processors sitting at a batch root are finally caught. - for_each and lane_consumer compiled their sub-chains without the enclosing scope's Strict flag, so even opted-in strict compiles only warned inside record sub-chains. New ScopeAwareProcessor interface: after construction the compile hands hosts its ScopeInfo and they recompile their sub-chain strict (if/else and case already threaded theirs through WithScope constructors). Tests: strict compile rejects a batch-scoped processor inside for_each (instance-declared and registry-declared variants); non-strict stays warn-only; ValidatePipelineConfig surfaces the nested mismatch in its warnings — the signal the pipeline-audit pre-flight consumes.
…68 #75) AcquireEvent/ReleaseEvent recycle *Event across record-mode batches: maps are cleared in place under an entry cap, double release is rejected atomically, and ReleaseEvent no-ops on codec/clone-allocated events so for_each can release every record unconditionally on all exit paths (including drops and failures). The P4 bench suite lands the gates: 1MB document K=4 record chain (pooled vs fresh), and the 1KB CPU-chain parallel gate (parallel:4 measures 3.8x on the 8-core M2, clearing the >=3x target). Also fixes a pre-existing -race failure that blocked the gate: debug logs passed the mutable *Processors to seelog's lazy async formatter; eagerly materialize procs.String() at the three call sites.
…he reflection walkers
Event.pooled was *pooledEvent while pooledEvent embeds Event: a recursive
type that sent util.GetTagsByTagName (no cycle guard) into infinite
recursion during model schema registration — coco crashed at startup
with a stack overflow. Hold the back-reference as interface{} (the
walker does not descend interface fields); ReleaseEvent type-asserts.
…r registry (#91) Chain generation is product logic (LLM backend, prompts, catalog); the endpoint is not. Products register a ChainGenerator at init; the unified endpoint validates the request shape (documents must be JSON objects, mode generate|refine, refine requires current_chain), resolves the generator (empty name resolves the single registration, refuses when ambiguous), returns 501 when no generator exists, 502 on generator failure. Validation/replay stays with the caller's /pipeline/dry-run follow-up.
…ositions Presentation-state-only mapping persisted alongside the config; keys use the "n<index>" convention because ucfg drops purely numeric map keys (reads them as array indices). Not indexed, omitempty for legacy configs.
…ssary (P8 #118) - references/lane.md: in-process record channel — capacity/overflow policies (block/drop/spill and their hidden costs), dynamic lanes, lane_consumer, unified output step, cross-lane commit correctness (durable-first / ack receipts / inline), and the queue-vs-lane selection guide distilled from the design spec's appendix. - references/glossary.md: canonical terminology for message/batch/ record, scopes, contracts, channels, and the control plane. - README: compact terminology table linking to the full glossary. - references index lists both pages. Also fixes the docs build under Hugo >= 0.139: the release-notes entry for #415 contained hugo's internal placeholder prefix verbatim (HAHUGOSHORTCODE-...), which the placeholder replacer misparses as an unterminated shortcode token ('illegal state in content; shortcode token missing end delim'). Reworded the changelog line. (The theme's .Site.IsMultiLingual deprecation also fails 0.139 builds; the vendored theme is gitignored, patched locally.) Verified: hugo 0.139.3 build exits 0, both new pages render.
…ins replay for real Dry-run replays bind one decoded record to the context and carry no message batch, so for_each's batch lookup missed and the host silently no-opped: every chain with a for_each step replayed as 'no effect', which hid the migration-equivalence question and made canvas dry-runs of wrapped chains meaningless. In dry-run mode with no batch and a bound record, for_each now adopts that record as its single input (the zero-decode typed entry lane_consumer uses at runtime) and runs the sub-chain stage-wise on it. Adopted records: serial execution (deterministic diffs), no encode pass, no pool release — the harness owns the record. Outside dry-run the historical no-batch skip stands. Process is restructured around runOnRecords(records, msgs) shared by the batch path and the adoption path. Tests: adoption mutates and restores the bound record; non-dry-run no-batch stays a no-op; full for_each suite green.
EnableDryRun stored its state in the pipeline Context's parameter map, which is invisible to code holding a context DERIVED from ctx.Context — exactly what processor helpers do (utils.EachDocumentRecord callbacks hand down plain contexts), so deep layers (orm, attachment stores) could never see dry-run mode. The state now lives in a context.WithValue chain installed on ctx.Context; InDryRun/DryRunCapture read through it, and new InDryRunCtx/DryRunCaptureCtx accept plain contexts for layers below processors. AcquireContext builds fresh Contexts, so nothing leaks across acquisitions. DryRunWrite.Kind comment widened (attachment captures coming from coco's blob-store guard). Framework suites green; coco's attachment dry-run guard (next commit) consumes DryRunCaptureCtx.
…131) PUT/POST/DELETE /pipelines used to only persist: a locally running task kept its stale compiled chain until the instance restarted. The CRUD handlers now trigger a rebuild that mirrors the fsnotify reload path — drain the in-flight batch (stopAndWaitForRelease), drop the task, and recreate from the new config. Matching is strict on task ID, so deployment-rendered tasks (which keep the ORM pipeline ID) reload while unrelated local YAML tasks stay untouched; transient tasks are skipped. A per-task generation counter (1 = the startup version) rides the runtime snapshot so controllers can observe that a reload landed. Rebuilds run off the request path and are serialized — rapid successive edits converge on the newest config instead of racing LoadOrStore.
…ordon, lock-takeover events (#120 #121 #122) #120 Control ops (_pause/_resume/_start/_stop and the new _drain) accept an optional ?generation= guard: an instruction applies only when the local task runs exactly that config revision, so an instance still on an older generation (hot reload not landed) never acts on instructions issued for the new one, and vice versa. Mismatches answer 409 carrying both generations; an absent param keeps today's unconditional behavior. #121 Graceful drain: a Drain flag on the pipeline context tells consumers to stop FETCHING — the in-flight batch finishes, its offset commits, then the worker returns (as opposed to CancelTask, which aborts mid-batch for redelivery). Endpoints: POST /pipeline/task/:id/_drain (parks the task STOPPED, resumable) and POST /pipeline/_drain (drain every task, one bounded shared wait). Cordon: POST /pipeline/_cordon / _uncordon (status via GET) — a cordoned instance lets running tasks finish but parks every start path (auto_start, _start, _resume, hot- reload recreations) until uncordoned. Restart clears the drain flag so a restarted consumer does not immediately re-drain. #122 Singleton lock takeovers are now visible: before re-acquiring an expired foreign lease, the task loop probes the previous owner and files a lock-takeover StepEvent (new Outcome) into the flight ring — the alert surface operators watch — plus a warn log.
Multi-instance deployments need per-instance attribution before cluster views can sum rows: StepEvent gains InstanceID stamped at the emitStep choke point (control-plane events via recordPipelineEvent stamp it too), StageStats rows self-describe their emitting node, and the persisted PipelineStageStat rows map instance_id as a keyword so cross-instance queries group on it — generalizing the field logpilot's InstanceEPS already carried. Additive JSON; old consumers ignore the new field.
…als (#125/#130 groundwork) Aggregates the persisted PipelineStageStat snapshots: each instance's LATEST tick per stage is kept (older ticks superseded, never summed), counters sum across instances, latency becomes the invocation-weighted mean, and PerInstance detail ships alongside the totals so the aggregate stays auditable — cluster total must equal the sum of the parts, which is exactly the #130 acceptance. Optional ?pipeline= filter; input order is irrelevant (rows sorted newest-first internally).
…st, instance proxy (#125) _rollout: the RolloutSpec model finally does something. The reconciler gates managed-config writes through rolloutPlan — canary wave first, the next batch only opens when every wave instance REPORTS running the staged content, completion when the waves cover the fleet. Readiness is honest: RuntimeSnapshot now emits the running task's content digest (same canonical six-field shape the deployment renderer produces), it rides _sync as config_digest, and the gate compares pushed vs reported — a STARTED-but-stale canary holds the rollout instead of green-lighting it. POST /pipeline/deployments/:id/_rollout {advance|rollback} overrides the machine: advance forces the next batch past readiness, rollback rewrites every target to the recorded pre-rollout content at once (a brake — the reconciler re-stages desired from canary; the response says so and points at pipeline-version rollback for permanence). Cluster broadcast: POST /pipeline/deployments/:id/_pause|_resume|_stop fans the command out to the deployment's targets THROUGH the reverse channel (managed agents are behind NAT; the agent-dialed websocket is the only path). Commands address the task by pipeline NAME and carry the caller's ?generation=N verbatim — instances on another config generation answer 409 (#120), so the aggregate shows exactly who refused and why. Instance ops: POST /pipeline/instances/:id/_drain|_cordon|_uncordon forward #121's instance-local primitives with query passthrough (timeout_seconds), plus GET /pipeline/instances/:id/proxy?path= as a read-only /pipeline/-scoped debug passthrough (allowlisted, traversal refused, instance status codes pass through verbatim). DeploymentStatus now mirrors rollout progress for the overview UI. Handler tests stub the reverse channel at a package seam; rollout tests cover the pure gate (canary/batch/hold/force/prune/complete), digest parity between render and instance-side reconstruction, and a full staged converge-hold-advance-complete walk over the sqlite harness.
…estReconcile poke - RuntimeReport.Cordoned: the pipeline module exposes the local cordon flag via configs_client.SetInstanceFlagsProvider and merges it into every runtime snapshot; the configs server stamps it onto Instance (placement keeps flowing configs so running tasks drain in place) - InstanceReport.Cordoned surfaces the flag in deployment status, so the fleet UI can show cordoned instances on the deployment matrix - RequestReconcile() pokes the deployment reconciler for an immediate pass so publishers converge right after upserting deployments instead of waiting out the 15s tick
create/update/delete of a deployment requests an immediate reconcile pass so convergence starts right away instead of on the next 15s tick — same behavior the logpilot publisher gets via RequestReconcile.
The keep_running loop parks the context in FINISHED during the inter-pass retry sleep. The runtime snapshot mapped FINISHED straight to STOPPED, so a healthy task flapped the fleet view to down once per second and rollout readiness (requires STARTED) stalled forever — caught live in #130 acceptance with two gateways both running the pinned echo pipeline while the deployment reported 0/2 ready. FINISHED + keep_running + not exiting now reports STARTED (the loop goroutine is alive and will run again); FAILED passes still report FAILED so crash-loops hold rollout waves. STOPPING also maps to STOPPED instead of UNKNOWN.
…iness requires a live heartbeat LastSeen stamped time.Now() at read time, so a killed instance kept displaying STARTED with a brand-new timestamp forever — seen live in #130 when the lock-holding gateway was killed: the dead row looked fresher than the takeover. Now it shows the instance's last _sync heartbeat. Rollout readiness additionally requires a heartbeat within 3 default sync intervals: a frozen STARTED+digest report from a dead instance must not keep counting as a ready canary.
…er registry The gateway registers this module for one documented purpose — receiving LogPilot's sink-cluster pushes on /easysearch/ at stream publish time — but with no easysearch: section in its yml the module stayed dark and publishes failed with bare 404s while the sink lane logged 'cluster metadata not ready' forever (found live in #130 acceptance). EnabledByDefault flips the module on when the section is absent, so the registering app's intent survives an unconfigured yml.
…y-enabled apps Two live-verification bugs from #129 product binding: - installStats registered one statsRecorder as the StepEvent observer and stored a SECOND instance in globalStats — counters accumulated into an object nobody read; every stage view (runtime endpoint, persistence, cluster aggregation) stayed empty forever. One instance now serves both roles; TestStats_ObserverWiring drives a real Processors.Process chain and fails on the two-instance shape. - persistStageSnapshot wrote with a bare orm context; the security data-op hook requires a session user or the direct-access flags, so the first snapshot tick panicked ("invalid user") and killed security-enabled apps (coco) the moment pipeline_metrics persistence was enabled. The writer is a background system writer — it now uses DirectAccess().
…bounds the table Instances without a shared datastore (gateway-type, local sqlite) never reached the manager's cluster aggregation — /pipeline/cluster/stages only saw rows from ES-primary apps. The counters now travel the existing heartbeat: - RuntimeReport grows stage_stats: the instance's cumulative per-stage rows (wire shape common.StageStat), collected by the pipeline module (reportedStageRows — processor rows tagged by pipeline Config.Name, lane carriers once, not per pipeline) and attached by the sync client - the configs server sinks reported rows into its own store via PersistReportedStageStats (registered from PipeModule.Setup; inert on apps without the configs server) — same row shape and latest-tick fold as the local exporter - pipeline_metrics.persistence.retention_seconds (default 1h, negative disables): both write paths prune beyond the window — the fold only reads each instance's latest tick, unbounded history would eventually crowd the aggregation query - test ORM setup refactored into a side-effect-free ensureDeployORMHandler: orm.Register is a process-global last-wins registry, a second handler registration anywhere flips later tests onto a handler missing their schemas Live-verified on the e2e fleet: gateway's 边缘心跳 echo/node_stamp rows land in coco's store via _sync (two cumulative ticks) and the product's cluster tab aggregates them beside coco's own connector_dispatcher rows.
…lector The placement selector matched instance labels only, but fleets are organized by console-assigned groups (server-owned data instances never report as labels), so deployments could not target a group. The reserved `group` selector key now matches membership instead: values are comma-separated group names, any of which puts the instance in scope; other keys keep plain label equality. Labels and groups are separate namespaces — a label named "group" does not satisfy the group key. Membership is re-evaluated every reconcile cycle, so instances added to a group pick the deployment up automatically. Docs updated.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
The October pipeline refactoring branch: a unified pipeline engine with record contracts and validation gates, distributed deployment management, and a new memory-path bulk indexing pipeline built on lanes. 31 commits.
Unified pipeline engine
527a7d90)a05e5756)case/branchesmulti-way dispatch processorDistributed deployment management
_syncLane-based bulk indexing (memory-path twin of bulk_indexing)
merge_to_bulk_lane(merge records onto a lane) andbulk_indexing_lane(subscribe and ship payloads to Elasticsearch with batch receipt settle)lane.GetOrCreatereplaces closed lanes — consumers closing their lane on exit no longer wedge producers foreverFixes
Docs
Processor reference pages (bulk_indexing_lane, merge_to_bulk_lane, lane_consumer, if, output, tap), pipeline validation gate and deployment API references.
Testing
All affected packages pass: core (lane/pipeline/otel/util), modules (pipeline/queue/configs/elastic/stats), plugins (indexing_merge, lane_bulk);
go test -raceclean on core/lane and plugins/elastic/lane_bulk; full non-enterprise tree builds. Note:modules/sqlitebackup work (#432) and the core/taskInitialDelay/state-lock changes are already on main via that PR and are merged into this branch.