SY-4333: Move Signal Propagation Out of Distribution Layer

[pjdotson]

Issue Pull Request

Linear Issue

SY-4333

Description

Moves change-data-capture (CDC) signal propagation out of the distribution layer and into the service layer, where it belongs architecturally. After this change the distribution layer has no knowledge of signals at all.

Package move

• The signals package moves from core/pkg/distribution/signals to core/pkg/service/signals.
• The per-entity CDC publisher subpackages move alongside it: distribution/{channel,group,ontology}/signals → service/{channel,group,ontology}/signals.

Layer wiring

• distribution/layer.go no longer constructs the signals.Provider, runs any CDC publishers, or exposes a Signals field, and the EnableServiceSignals config knob is removed.
• service/layer.go now owns provider construction and starts the channel, group, and ontology publishers on open. All downstream services consume l.Signals instead of cfg.Distribution.Signals.
• The ontology CDC publisher is no longer gated to the bootstrapper node — it now publishes on every node, matching how label, status, channel, and group signals already behave.

Layering fix

• The distribution test mock (distribution/mock/cluster.go) no longer exposes a signals.Provider or imports service/signals, removing a distribution→service layering inversion. Service-layer tests that need a provider construct one directly from the mock node's Channel/Framer.

Ontology relationship CDC simplification

• service/ontology/signals now publishes relationship changes straight from the ontology's relationship gorp table via signals.PublishFromGorp (the same pattern group signals use), replacing a hand-written observable translator. The set/delete wire format is unchanged (the relationship key string, which round-trips via ParseRelationship). Resource CDC stays on the aggregate resource observer, since resources are sourced from every registered ontology service rather than a single gorp table.

Tests

• Adds CDC propagation test suites for service/channel/signals and service/group/signals.
• Signals test suites now provision in BeforeSuite and no longer rely on Ordered containers.

Basic Readiness

• I have performed a self-review of my code.
• I have added relevant, automated tests to cover the changes.
• I have updated documentation to reflect the changes.

Greptile Summary

This PR moves CDC (change-data-capture) signal propagation from the distribution layer into the service layer, resolving a layering inversion where distribution/signals had knowledge of service-level concerns. The signals.Provider is now constructed in service/layer.go and distributed downward instead of up from distribution/layer.go.

• Package migration: distribution/signals → service/signals; per-entity publishers for channel, group, and ontology move alongside it. distribution/layer.go no longer carries a Signals field or EnableServiceSignals config knob.
• Signal publishing centralised in layer.go: rack, task, arc, label, and status CDC publishers are now wired directly in OpenLayer; services that have custom CDC logic (user, rbac, ranger, etc.) still accept a *signals.Provider from the layer instead of from cfg.Distribution.Signals.
• Ontology bootstrapper gate removed: ontology resource/relationship signals now publish on every node, matching the existing behaviour of channel, group, label, and status signals.

Confidence Score: 5/5

This is a clean architectural refactoring with no behaviour changes to the CDC wire format or channel names. Safe to merge.

The channel names produced by DefaultGorpPublisherConfig are preserved identically for all entry types (Status[any] has a CustomTypeName implementation returning Status; all other types are non-generic). The signal publisher pipelines are functionally identical to what was in the distribution layer. Service initialization order in layer.go is correct, and all Observe() hooks are present on their respective services.

No files require special attention. The most complex change is the new GorpPublisherConfig generic machinery in service/signals/gorp.go, which is well-validated and covered by existing tests.

Important Files Changed

Filename	Overview
core/pkg/service/layer.go	Central orchestration file where all signal publisher wiring now lives; initialization order is correct and all Observe() methods are present on the services they're called on.
core/pkg/service/signals/gorp.go	New generic GorpPublisherConfig and helpers; OR-based merge of DisableSet/DisableDelete is intentional, DefaultGorpPublisherConfig name inference works correctly for all entry types used.
core/pkg/service/signals/publisher.go	PublishFromObservable constructs a plumber pipeline; cfg mutation for channel key back-fill is safe (local copy), and series with empty DataType when a channel is disabled are never written to the frame.
core/pkg/service/ontology/signals/signals.go	New service-layer ontology signals; resource CDC uses the aggregate observer, relationship CDC uses PublishFromGorp - a cleaner approach than the removed hand-written translator.
core/pkg/distribution/layer.go	Correctly strips Signals field, EnableServiceSignals config, and all CDC publisher wiring; the distribution layer now has zero knowledge of signals.
core/pkg/service/channel/signals/signals.go	New package wrapping GorpPublisherConfig for channel CDC; custom MarshalSet serializes via ToPayload+JSON+MarshalVariableSample, consistent with the old distribution-layer implementation.
core/pkg/service/rack/service.go	Signals field removed and replaced by the new Observe() hook; gorp.OpenTable and gorp.NewEntryMigration calls simplified to infer type parameters.
core/pkg/service/framer/service.go	Adds framer.Wrap for lightweight test setup; comment accurately documents that calculation-backed paths are not wired up.
core/pkg/distribution/mock/cluster.go	Removes EnableServiceSignals: new(false), eliminating the distribution to service layering inversion in mock setup.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    subgraph ServiceLayer["service/layer.go (OpenLayer)"]
        SL_Signals["signals.Provider (NEW)"]
        SL_CDC["CDC Publishers (NEW)"]
        SL_Other["User / RBAC / Ranger / ... Services"]
    end

    subgraph DistLayer["distribution/layer.go"]
        DL_Channel["Channel"]
        DL_Group["Group"]
        DL_Ontology["Ontology"]
    end

    subgraph CDCPubs["CDC Publishers"]
        P1["sy_channel_set/delete"]
        P2["sy_group_set/delete"]
        P3["sy_ontology_resource/relationship"]
        P4["sy_label_set/delete"]
        P5["sy_status_set/delete"]
        P6["sy_rack_set/delete"]
        P7["sy_task_set/delete"]
        P8["sy_arc_set/delete"]
    end

    SL_Signals --> SL_CDC
    SL_CDC --> CDCPubs
    SL_Signals --> SL_Other
    DL_Channel --> P1
    DL_Group --> P2
    DL_Ontology --> P3
    CDCPubs -.->|stream| FreeChannels["Free Virtual Channels"]

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
    subgraph ServiceLayer["service/layer.go (OpenLayer)"]
        SL_Signals["signals.Provider (NEW)"]
        SL_CDC["CDC Publishers (NEW)"]
        SL_Other["User / RBAC / Ranger / ... Services"]
    end

    subgraph DistLayer["distribution/layer.go"]
        DL_Channel["Channel"]
        DL_Group["Group"]
        DL_Ontology["Ontology"]
    end

    subgraph CDCPubs["CDC Publishers"]
        P1["sy_channel_set/delete"]
        P2["sy_group_set/delete"]
        P3["sy_ontology_resource/relationship"]
        P4["sy_label_set/delete"]
        P5["sy_status_set/delete"]
        P6["sy_rack_set/delete"]
        P7["sy_task_set/delete"]
        P8["sy_arc_set/delete"]
    end

    SL_Signals --> SL_CDC
    SL_CDC --> CDCPubs
    SL_Signals --> SL_Other
    DL_Channel --> P1
    DL_Group --> P2
    DL_Ontology --> P3
    CDCPubs -.->|stream| FreeChannels["Free Virtual Channels"]

Loading

_{Reviews (4): Last reviewed commit: "Merge branch 'rc' into sy-4333-move-sign..." | Re-trigger Greptile}

[codecov]

Codecov Report

❌ Patch coverage is 76.37363% with 43 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.77%. Comparing base (26b7a31) to head (194a5d9).

Files with missing lines	Patch %	Lines
core/pkg/service/signals/gorp.go	42.10%	22 Missing ⚠️
core/pkg/service/ontology/signals/signals.go	84.84%	7 Missing and 3 partials ⚠️
core/pkg/service/signals/publisher.go	91.93%	2 Missing and 3 partials ⚠️
core/pkg/service/arc/service.go	0.00%	2 Missing ⚠️
core/pkg/service/label/service.go	50.00%	2 Missing ⚠️
core/pkg/service/rack/service.go	50.00%	2 Missing ⚠️

❌ Your patch check has failed because the patch coverage (76.37%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage.

Additional details and impacted files

@@            Coverage Diff             @@
##               rc    #2494      +/-   ##
==========================================
+ Coverage   67.67%   67.77%   +0.10%     
==========================================
  Files        2546     2565      +19     
  Lines      124876   126780    +1904     
  Branches     9153     9164      +11     
==========================================
+ Hits        84507    85929    +1422     
- Misses      33833    34141     +308     
- Partials     6536     6710     +174     

Flag	Coverage Δ
client-py	86.32% <ø> (ø)
client-ts	91.72% <ø> (+<0.01%)	⬆️
console	31.44% <ø> (ø)
core	71.90% <76.37%> (-0.07%)	⬇️
pluto	61.15% <ø> (+0.04%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[pjdotson]

@greptile /rerun

[pjdotson]

@greptile /rerun

[emilbon99]

Should this still be a sub-package? The primary reason I had to put these in a separate package was to avoid circular deps. That isn't necessary anymore, so why isn't this core/pkg/service/ontolgoy.PublishSignals?

[pjdotson]

see below

[emilbon99]

So core/pkg/service/signals only depends on the distribution layer channels service. That means that there won't be a circular dep issue, and this package can get moved directly into the service layers channel service right?

[pjdotson]

Changed it so that the signals service imports channel.Service and framer.Service. will be moving most channel functionality out of the distribution layer into the service layer channel service and the ontology will eventually be in core/pkg/service/ontology as well.

[pjdotson]

@greptile /rerun