Relay - #175
Closed
johankristianss wants to merge 24 commits into
Closed
Conversation
- Add ping handler with read deadline for stale connection detection - Use context-based HTTP requests cancelled on tunnel disconnect - Remove fixed 120s client timeout in favor of context cancellation
- Initialize nodesMap in calcNodes() to prevent race when shallow-copied ProcessGraph instances share the same underlying map - Add concurrent ToJSON test to verify the fix under -race - Skip already unregistered executors in stale executor cleanup - Bump version to v1.9.13-beta4 - Switch docker-compose defaults to embedded DB
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #175 +/- ##
==========================================
- Coverage 73.07% 70.67% -2.40%
==========================================
Files 265 280 +15
Lines 23917 25612 +1695
==========================================
+ Hits 17477 18102 +625
- Misses 4467 5206 +739
- Partials 1973 2304 +331 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
AddExecutor and RemoveExecutorByName had a TOCTOU race: the check-then-act on executor existence was not atomic, allowing concurrent calls to corrupt indexes. Protect both operations with executorMu. Also adds the executorMu field to the EmbeddedDatabase struct and a concurrent re-registration test.
Flexible key-value metrics system for executors. Supports GAUGE (overwrite) and COUNTER (increment) types with optional period bucketing (day/week/month) for time-series tracking. - Core model with period-aware ID generation - Embedded DB and PostgreSQL implementations - RPC messages, handlers with executor ownership auth - Client SDK and CLI commands (metrics ls/get/history/rm) - Comprehensive tests for both DB backends
Replace ad-hoc per-entity locks with a single db.mu sync.RWMutex. All write methods acquire db.mu.Lock(), reads use store-level locks. Methods called internally from other locked methods are split into public (locked) + internal (unlocked) versions to avoid deadlock. - Batched retention policy to avoid long lock holds - 7 simulation stress tests (LLM fleet, burst assignment, deadlock detector) all passing with -race - Architecture README documenting WAL, stores, indexing, concurrency
- Change etcd test ports from 2379/2380 to 12379/12380 to avoid conflicts with running colonies server - Add MetricDatabase methods to controllers DatabaseMock
Required for browser-based ColonyFS file downloads via /api/fs/ endpoint. Without this, CORS preflight rejects the custom auth headers.
Fixes "websocket: close 1006 (abnormal closure): unexpected EOF" server errors. The client now sends a CloseNormalClosure frame before closing the connection.
Catch SIGINT/SIGTERM in the server start command to flush the embedded DB before exit. Without this, WAL-buffered state changes (e.g. executor UNREGISTERED) could be lost on Ctrl+C, causing stale registrations to reappear after restart. - Signal handler calls srv.Shutdown() then db.Close() - Remove retry loop so ServeForever returns cleanly after shutdown - Add shutdown persistence tests verifying state survives restart
Server-side: replace HandleHTTPError with sendWSErrorMsg in the WebSocket handler. After upgrading to WebSocket, HTTP error responses corrupt the connection framing, causing "unexpected EOF" on subsequent connections. Client-side: send proper close frame (code 1000) before closing WebSocket connections. Exit read goroutines on error instead of looping. Add debug logging to WebSocket subscription lifecycle.
New RPC endpoint that adds a child process to a process graph without blocking it on the parent's completion. The child runs immediately while remaining visible in the DAG for traceability. Use case: agentic tool loops that submit tools sequentially but need them visible in the workflow graph.
AddChild with insert=true now skips independent children when reparenting. Independent children stay attached to their original parent, while dependent children are moved to the inserted node. Also adds Independent flag to Process struct for tracking.
- RootFunc stored on ProcessGraph, set during graph creation - PostgreSQL: ROOT_FUNC column, filtered in WHERE clause - Embedded DB: filtered during index traversal - New FindProcessGraphsByState with excludeRootFuncs parameter - RPC: ExcludeRootFuncs field on GetProcessGraphsMsg - Client: GetProcessGraphsByState with exclude parameter - Handler routes to filtered query when excludeRootFuncs present
Support excludeRootFuncs parameter to filter out process graphs by root function name. Also set RootFunc from NodeName when FuncName is empty. Adds tests for both embedded and PostgreSQL backends.
Two bugs caused crons to stop firing permanently: 1. resolveInitiator failed with "Could not derive InitiatorName" when the executor that created the cron was re-registered with a new ID. Now falls back to using the raw ID as the initiator name instead of erroring. 2. StartCron did not advance NextRun when CreateProcessGraph failed, causing the cron to retry every second forever. Now advances NextRun even on failure.
Two latency fixes for dynamic agentic workflows that submit processes with Conditions.ExecutorNames (single named recipient) and use AddChild to mutate the graph mid-execution. 1. eventhandler: broadcast to all type-matching listeners when a process is routed (ExecutorNames set), instead of round-robin waking exactly one. With N executors of the same type but only 1 valid target, the round-robin lands on the right executor only ~1/N of the time. The wrong picks instantly fail Assign() (name mismatch) and re-poll. The correct executor sits in its 10s long-poll until either timeout or a future round-robin happens to land on it — observed as a 5-9s queue spike per routed tool call. Broadcasting wakes all candidates; only the named one wins the assign atomically; the rest no-op (cheap). Round-robin thundering-herd protection is preserved for unrouted processes (the bulk-throughput case). 2. controllers: replace the flat 500ms × 10 retry loop in Resolve with exponential backoff (1ms → 50ms cap, 50 retries, ~1s total budget). The retry handles same-server graph-mutation races (AddChild's row not yet visible when a concurrent Assign tries to Resolve) and multi- server propagation delays. The original 500ms sleep added 5s worst- case wall-clock per assignment under dynamic-graph load. Same correctness guarantee, ~50-100x lower typical latency. Plus diagnostic logging in HandleAssignProcess (per-call AssignDurMs, TryCount, WaitEventCount, WaitFromSubMs) and resolveWithBackoff (retry count + duration) to make this whole subsystem observable. Without these the 5-9s spikes were impossible to attribute.
Companion primitive to SubscribeProcess / SubscribeProcesses. Lets clients subscribe to file events (added / updated / removed) under a ColonyFS label or label prefix instead of polling GetFileData. Motivating use case: event-driven workflows that need to react when a file lands in a label. Today consumers either poll or invent sentinel-process workarounds. A first-class subscription closes the gap and reuses the existing realtime websocket plumbing — small marginal cost on the server side. This PLAN.md is design only, no code yet. Covers wire format (SubscribeFilesPayloadType), client API shape (Go + TS), server-side dispatch via the existing realtime backend, filtering rules (label-prefix + kind), backpressure (drop oldest with a lag marker), test plan, four-phase rollout, and open questions on revision semantics, batch delivery, and path matching syntax.
Phase 1 of the file pubsub feature (PLAN.md committed earlier in the
relay branch). Adds the foundational types with full unit-test
coverage; no server- or client-side wiring yet.
pkg/rpc/subscribe_files_msg.go:
- SubscribeFilesPayloadType constant ("subscribefilesmsg")
- SubscribeFilesMsg with ColonyName, LabelPrefix, Kinds, Timeout,
MsgType. Constructor, ToJSON / ToJSONIndent, Equals (including
element-wise Kinds compare), CreateFromJSON.
- Tests cover round-trip, indented round-trip, Equals across each
field as the discriminator, MsgType drift, identical-Kinds-with-
different-backing-arrays, empty/nil Kinds sentinel, payload-type
pinning, and JSON tag shape.
pkg/core/file_event.go:
- FileEventKind iota: FileAdded(1), FileUpdated(2), FileRemoved(3).
Wire-stable integers; do not renumber.
- FileEvent type with Kind, ColonyName, Label, Name, FileID, Size,
Checksum, ChecksumAlg, Timestamp. JSON-marshalled with
omitempty on the file-identifying fields so FileRemoved events
don't carry a misleading FileID="" or Size=0.
- MatchesPrefix and MatchesKinds — the matchers the server will use
on the fan-out path. Prefix uses a path-shaped rule (exact match
OR HasPrefix(prefix + "/")), so /home/root/inbox/2026 matches a
subscription on /home/root/inbox but /home/root/in does not match
/home/root/inbox.
- Constructors CreateFileAddedEvent / CreateFileUpdatedEvent /
CreateFileRemovedEvent (the last takes coordinates rather than a
*File pointer because the file is already gone by the time the
caller invokes it).
- Tests cover wire-stable enum values, string formatting, JSON
round-trip, omitempty behaviour for FileRemoved, the matcher edge
cases (empty prefix, exact, descendant-with-separator, sibling,
non-separator-prefix, kind filter with garbage values), every
constructor, Equals including timestamp non-equality, and JSON
tag shape.
Phase 2 will add the server-side subscription registry and the publish
hooks in pkg/server/handlers/file/. Phase 3 the client API plus
end-to-end integration tests.
Phase 2a of the file pubsub feature: the in-process event bus that publishes file events to interested subscribers. Pure logic, no websocket coupling — that lands in the next commit. pkg/backends/file_event_bus.go: - FileEventBus interface: Publish, Subscribe, NumberOfSubscribers, Stop. Single-server scope; no cross-server replication (matches the process-subscription model already in this repo). - inMemoryFileEventBus default implementation. Per-subscriber buffered channel (DefaultEventBufferSize=256). Drop-oldest on overflow with ErrSubscriberOverflowed delivered exactly once per overflow streak on a separate error channel; re-armed automatically once the subscriber drains. - Filtering: same colony, MatchesPrefix on the label, MatchesKinds on the event kind. Kinds slice is defensively copied at subscribe time so caller-side mutation doesn't drift the live filter. - Lifecycle: ctx cancel removes the subscription and closes both channels; Stop drains every active subscriber and turns Publish into a no-op. Idempotent. - Concurrency: bus.mu (RWMutex) held through the whole Publish fan-out, so removeSubscriber waits for in-flight sends to finish before closing the subscriber's channels. Sends are non-blocking per subscriber, so a slow subscriber can't stall fast ones. pkg/backends/file_event_bus_test.go (15 tests, all -race-clean): - Happy-path delivery; matchers (colony / label-prefix / kind) each verified independently with both positive and negative cases. - Empty-prefix wildcard semantics with explicit cross-colony isolation guard. - Multi-subscriber fan-out where filters differ. - Context-cancel cleanup (channels close, NumberOfSubscribers drops back to zero). - Stop idempotency, post-Stop Publish/Subscribe behaviour. - Backpressure: 50-event burst into a 2-slot buffer triggers overflow signal and bounds the delivered count to <= buffer+1. - Buffer-size clamping for non-positive args. - Nil-event tolerance. - Per-colony NumberOfSubscribers counts. - TestConcurrentPublishAndSubscribe: 4 publisher goroutines x 200 events x 8 subscriber goroutines x 50 subscribe/cancel cycles. Catches the close-then-send race that the per-fan-out lock fixes. - Defensive-copy guard for the kinds slice. Phase 2b will hook this bus into the file handlers (publish on HandleAddFile / HandleRemoveFile success). Phase 2c wires the websocket subscription dispatch.
Phase 2b: hook the file handlers into the FileEventBus. The handler publishes after the underlying DB call succeeds. The bus is optional (nil-tolerant) so deployments / tests that haven't enabled realtime keep working unchanged. Discrimination between FileAdded and FileUpdated: we look up GetLatestFileByName(colony, label, name) BEFORE the AddFile call. If a prior revision exists, this is an update; otherwise a new file. A transient lookup error degrades to FileAdded rather than blocking the write — strictly best-effort signalling. HandleRemoveFile by-ID resolves the file's label+name BEFORE deleting so the published FileRemoved event carries the right coordinates; RemoveFile by-name uses the kwargs directly. Either way, FileRemoved omits FileID/Size/Checksum (they're misleading after the delete) via omitempty on the FileEvent struct. Plumbing: - file.Server interface gains FileEventBus() backends.FileEventBus. - ServerAdapter delegates to *Server.FileEventBus(). - *Server gets fileEventBus field + FileEventBus() / SetFileEventBus / EnableFileEventBus accessors. Lazy: zero overhead when nobody has subscribed. Tests (handlers_pubsub_test.go, 9 cases, all -race-clean): - HandleAddFile publishes FileAdded with full metadata for first revisions. - HandleAddFile publishes FileUpdated for new revisions of an existing (colony, label, name). - HandleRemoveFile (by name) publishes FileRemoved with right coordinates and no file-id/checksum. - HandleRemoveFile (by id) does the lookup-before-delete so the event still has label+name even though the caller only had the id. - Nil bus is a clean no-op for both handlers. - Cross-colony isolation: events for colony A don't leak to a colony-B subscriber. - Label-prefix scoping survives end-to-end through the handler: /home/root/inbox subscriber sees /home/root/inbox/2026 events but not sibling /home/root/outbox events. - Failed AddFile (membership rejected) does NOT publish — events only fire after the DB write succeeds. Phase 2c (next commit) wires the websocket subscription dispatch: new SubscribeFiles payload routing in pkg/backends/gin/realtime.go, file subscription adapter in pkg/backends/gin/, end-to-end test that opens a websocket, subscribes, writes a file, observes the event.
Phase 2c + 3 of the file pubsub feature: closes the loop end-to-end. pkg/backends/gin/realtime.go: - RealtimeServer interface gains FileEventBus(); ServerAdapter already exposes it from the earlier publish-hook commit. - HandleWSRequest dispatches rpc.SubscribeFilesPayloadType to the new handleSubscribeFiles handler. - handleSubscribeFiles parses the SubscribeFilesMsg, verifies colony membership via Validator.RequireMembership (same auth as AddFile / GetFile), gets a bus subscription, and pumps events to the websocket inside an RPCReplyMsg envelope. Bypasses the colonies-controller command queue (mirrors the ChannelRouter approach in handleSubscribeChannel) so a slow file-event subscriber never serialises unrelated colony work. - Overflow signals (ErrSubscriberOverflowed) are surfaced to the client as informational error messages but the subscription stays open so the client can resync via GetFileData. - Timeout 0 maps to a 24h server-side cap so a forgotten subscriber doesn't pin resources forever. pkg/client/subscription.go: - New FileSubscription type with EventChan, ErrChan, conn, Close(). pkg/client/realtime_client.go: - New SubscribeFiles(colonyName, labelPrefix, kinds, timeout, prvKey) method, mirroring SubscribeProcesses. Builds the RPC msg, opens the realtime websocket, and runs a read goroutine that decodes FileEvents into EventChan and surfaces errors on ErrChan. pkg/server/handlers/realtime/file_handler_test.go (7 E2E tests): - DeliversAddedEvent: full path subscribe -> AddFile -> client receives FileAdded with correct (colony, label, name, fileid, size, checksum). - DeliversUpdatedEvent: second revision of same name produces FileUpdated, not a duplicate FileAdded. - KindFilter: subscribing to FileAdded only suppresses subsequent FileRemoved events. - LabelPrefixScoping: /inbox subscriber sees /inbox/2026 events but not /outbox events. - CrossColonyIsolation: colony-2 subscriber doesn't see colony-1 events. - RejectsForeignColony: subscribing with a key that doesn't own colony membership is rejected (either at handshake or via ErrChan). - MultipleSubscribers: wide and narrow filters in the same colony both receive the events that match each filter. All tests run against the real Colonies server stack (etcd, Postgres, gin websocket, full RPC pipeline) and pass. The feature is now end-to-end functional: clients can SubscribeFiles to a label prefix, write files via AddFile (or have any other client write them), and observe FileAdded / FileUpdated / FileRemoved events in real time over a websocket.
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
1 participant
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.
No description provided.