| 1 | package sidecar |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "log/slog" |
| 7 | "sort" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | |
| 11 | "reasonix/internal/extension" |
| 12 | "reasonix/internal/extension/protocol" |
| 13 | "reasonix/internal/pluginpkg" |
| 14 | "reasonix/internal/secrets" |
| 15 | ) |
| 16 | |
| 17 | const ( |
| 18 | // maxConcurrentPackageStarts bounds process creation while still preventing |
| 19 | // one slow optional runtime from serially delaying every package after it. |
| 20 | maxConcurrentPackageStarts = 4 |
| 21 | // packageStartupBudget is shared by every runtime in one generation. Without |
| 22 | // a generation-level budget, N stalled optional runtimes could delay boot by |
| 23 | // N times the per-client handshake timeout. |
| 24 | packageStartupBudget = defaultHandshakeTimeout |
| 25 | ) |
| 26 | |
| 27 | type clientStarter func(context.Context, ClientOptions) (*Client, error) |
| 28 | |
| 29 | type packageStartJob struct { |
| 30 | item pluginpkg.InstalledPackage |
| 31 | opts ClientOptions |
| 32 | } |
| 33 | |
| 34 | type packageStartResult struct { |
| 35 | client *Client |
| 36 | err error |
| 37 | } |
| 38 | |
| 39 | // RequiredStartError reports a required runtime package that failed to start |
| 40 | // or hand shake. It fails the whole build; an optional package's failure is a |
| 41 | // warning instead. errors.As distinguishes the two at the boot call site. |
| 42 | type RequiredStartError struct { |
| 43 | Plugin string |
| 44 | Err error |
| 45 | } |
| 46 | |
| 47 | func (e *RequiredStartError) Error() string { |
| 48 | return fmt.Sprintf("required extension runtime %q failed to start: %v", e.Plugin, e.Err) |
| 49 | } |
| 50 | |
| 51 | func (e *RequiredStartError) Unwrap() error { return e.Err } |
| 52 | |
| 53 | // LoadRuntimePackages returns the installed, ENABLED packages that declare a |
| 54 | // v1 runtime. This is the only enumeration the Manager ever launches from — |
| 55 | // see the package doc for the authorization invariant. |
| 56 | func LoadRuntimePackages(home string) ([]pluginpkg.InstalledPackage, []string) { |
| 57 | installed, warnings := pluginpkg.LoadInstalled(home) |
| 58 | var out []pluginpkg.InstalledPackage |
| 59 | for _, item := range installed { |
| 60 | if item.Package.Manifest.Runtime != nil { |
| 61 | out = append(out, item) |
| 62 | } |
| 63 | } |
| 64 | return out, warnings |
| 65 | } |
| 66 | |
| 67 | // Manager owns every sidecar started for one runtime generation. It is the |
| 68 | // ONLY sidecar launch path, and its inputs are the pluginpkg installed state |
| 69 | // alone. It implements io.Closer so the kernel's RuntimeSet can retire it |
| 70 | // with its controller generation. |
| 71 | type Manager struct { |
| 72 | mu sync.Mutex |
| 73 | clients map[string]*Client |
| 74 | closed bool |
| 75 | } |
| 76 | |
| 77 | // StartPackages loads the installed enabled v1 runtime packages from home and |
| 78 | // spawns each one, running the initialize handshake against sessionCtx. A |
| 79 | // package whose manifest marks the runtime Required fails the whole call: |
| 80 | // everything started so far is shut down and the error is a |
| 81 | // *RequiredStartError. An optional package's failure is collected as a |
| 82 | // warning string and startup continues. |
| 83 | // |
| 84 | // The ui handler serves every client's host/ui/* calls. A handler that also |
| 85 | // implements UIBinder (the stage-8 UI hub) receives a per-plugin binding and |
| 86 | // crash notifications instead of sharing one unbound handler. |
| 87 | func StartPackages(ctx context.Context, home string, sessionCtx protocol.SessionContext, ui UIHandler) (*Manager, []string, error) { |
| 88 | packages, warnings := LoadRuntimePackages(home) |
| 89 | startupCtx, cancel := context.WithTimeout(ctx, packageStartupBudget) |
| 90 | defer cancel() |
| 91 | m, runtimeWarnings, err := startLoadedPackages(startupCtx, packages, sessionCtx, ui, StartClient) |
| 92 | warnings = append(warnings, runtimeWarnings...) |
| 93 | return m, warnings, err |
| 94 | } |
| 95 | |
| 96 | // startLoadedPackages starts a previously discovered, deterministically ordered |
| 97 | // package set. Handler binding stays serial; process startup and handshakes use |
| 98 | // a bounded worker pool and the caller's shared generation context. Results are |
| 99 | // consumed in package order so warnings and required-failure selection do not |
| 100 | // depend on goroutine completion order. |
| 101 | func startLoadedPackages(ctx context.Context, packages []pluginpkg.InstalledPackage, sessionCtx protocol.SessionContext, ui UIHandler, start clientStarter) (*Manager, []string, error) { |
| 102 | m := &Manager{clients: make(map[string]*Client)} |
| 103 | if len(packages) == 0 { |
| 104 | return m, nil, nil |
| 105 | } |
| 106 | var binder UIBinder |
| 107 | if b, ok := ui.(UIBinder); ok { |
| 108 | binder = b |
| 109 | } |
| 110 | jobs := make([]packageStartJob, len(packages)) |
| 111 | for i, item := range packages { |
| 112 | pluginID := item.Installed.Name |
| 113 | clientUI := ui |
| 114 | if binder != nil { |
| 115 | clientUI = binder.HandlerFor(pluginID) |
| 116 | } |
| 117 | jobs[i] = packageStartJob{item: item, opts: ClientOptions{ |
| 118 | Package: item.Package, |
| 119 | Installed: item.Installed, |
| 120 | Session: sessionCtx, |
| 121 | UI: clientUI, |
| 122 | OnCrash: func(err error) { |
| 123 | slog.Warn("extension sidecar crashed", "plugin", pluginID, "err", secrets.RedactError(err)) |
| 124 | if binder != nil { |
| 125 | binder.ClientCrashed(pluginID) |
| 126 | } |
| 127 | }, |
| 128 | }} |
| 129 | } |
| 130 | |
| 131 | results := make([]packageStartResult, len(jobs)) |
| 132 | indices := make(chan int, len(jobs)) |
| 133 | for i := range jobs { |
| 134 | indices <- i |
| 135 | } |
| 136 | close(indices) |
| 137 | workers := min(maxConcurrentPackageStarts, len(jobs)) |
| 138 | var wg sync.WaitGroup |
| 139 | wg.Add(workers) |
| 140 | for range workers { |
| 141 | go func() { |
| 142 | defer wg.Done() |
| 143 | for i := range indices { |
| 144 | if err := ctx.Err(); err != nil { |
| 145 | results[i].err = fmt.Errorf("extension generation startup stopped before launch: %w", err) |
| 146 | continue |
| 147 | } |
| 148 | results[i].client, results[i].err = start(ctx, jobs[i].opts) |
| 149 | } |
| 150 | }() |
| 151 | } |
| 152 | wg.Wait() |
| 153 | |
| 154 | var warnings []string |
| 155 | var requiredErr *RequiredStartError |
| 156 | for i, result := range results { |
| 157 | item := jobs[i].item |
| 158 | pluginID := item.Installed.Name |
| 159 | if result.err != nil { |
| 160 | if item.Package.Manifest.Runtime.Required { |
| 161 | if requiredErr == nil { |
| 162 | requiredErr = &RequiredStartError{Plugin: pluginID, Err: result.err} |
| 163 | } |
| 164 | } else { |
| 165 | warnings = append(warnings, fmt.Sprintf("%s: optional extension runtime failed to start: %v", pluginID, result.err)) |
| 166 | } |
| 167 | continue |
| 168 | } |
| 169 | m.clients[pluginID] = result.client |
| 170 | } |
| 171 | if requiredErr != nil { |
| 172 | _ = m.Close() |
| 173 | return nil, warnings, requiredErr |
| 174 | } |
| 175 | return m, warnings, nil |
| 176 | } |
| 177 | |
| 178 | // Client returns the client for one plugin ID, or nil. |
| 179 | func (m *Manager) Client(pluginID string) *Client { |
| 180 | m.mu.Lock() |
| 181 | defer m.mu.Unlock() |
| 182 | return m.clients[pluginID] |
| 183 | } |
| 184 | |
| 185 | // Clients returns every live client ordered by plugin ID. |
| 186 | func (m *Manager) Clients() []*Client { |
| 187 | m.mu.Lock() |
| 188 | defer m.mu.Unlock() |
| 189 | out := make([]*Client, 0, len(m.clients)) |
| 190 | for _, client := range m.clients { |
| 191 | out = append(out, client) |
| 192 | } |
| 193 | sort.Slice(out, func(i, j int) bool { return out[i].pluginID < out[j].pluginID }) |
| 194 | return out |
| 195 | } |
| 196 | |
| 197 | // Close shuts every sidecar down in parallel; each client's own budgets |
| 198 | // bound the total. It is idempotent. |
| 199 | func (m *Manager) Close() error { |
| 200 | m.mu.Lock() |
| 201 | if m.closed { |
| 202 | m.mu.Unlock() |
| 203 | return nil |
| 204 | } |
| 205 | m.closed = true |
| 206 | clients := make([]*Client, 0, len(m.clients)) |
| 207 | for _, client := range m.clients { |
| 208 | clients = append(clients, client) |
| 209 | } |
| 210 | m.mu.Unlock() |
| 211 | |
| 212 | var wg sync.WaitGroup |
| 213 | for _, client := range clients { |
| 214 | wg.Add(1) |
| 215 | go func(c *Client) { |
| 216 | defer wg.Done() |
| 217 | _ = c.Close() |
| 218 | }(client) |
| 219 | } |
| 220 | wg.Wait() |
| 221 | return nil |
| 222 | } |
| 223 | |
| 224 | // Declaration-level kernel contribution payloads. They describe what a |
| 225 | // started sidecar declared; dispatch wiring arrives with stages 6-8. |
| 226 | |
| 227 | // InterceptorDecl is the KindInterceptor payload for one manifest-declared |
| 228 | // interceptor point. |
| 229 | type InterceptorDecl struct { |
| 230 | PluginID string |
| 231 | Point string |
| 232 | Priority int |
| 233 | } |
| 234 | |
| 235 | // StrategyDecl is the KindStrategy payload claiming one replacement slot. It |
| 236 | // implements the kernel's SlotClaimer so two runtimes claiming the same slot |
| 237 | // fail the build through ReplaceClaims. |
| 238 | type StrategyDecl struct { |
| 239 | PluginID string |
| 240 | Slots []extension.Slot |
| 241 | } |
| 242 | |
| 243 | // ReplacementSlots implements extension.SlotClaimer. |
| 244 | func (d StrategyDecl) ReplacementSlots() []extension.Slot { |
| 245 | return append([]extension.Slot(nil), d.Slots...) |
| 246 | } |
| 247 | |
| 248 | // ProviderDecl is the KindProvider payload for one handshake-declared |
| 249 | // extension-hosted provider. |
| 250 | type ProviderDecl struct { |
| 251 | PluginID string |
| 252 | Descriptor protocol.ProviderDescriptor |
| 253 | } |
| 254 | |
| 255 | // UIActionDecl is the KindUIAction payload for one handshake-declared action. |
| 256 | type UIActionDecl struct { |
| 257 | PluginID string |
| 258 | Decl protocol.UIActionDecl |
| 259 | } |
| 260 | |
| 261 | // Contributions renders every started client's declarations as kernel |
| 262 | // contributions: interceptor stubs per manifest intercept, strategy claims |
| 263 | // per manifest replaces, and provider / UI-action declarations from the |
| 264 | // validated handshake. Kernel-invalid IDs (whitespace, non-ref provider IDs) |
| 265 | // are skipped with a debug log rather than failing the whole snapshot. |
| 266 | func (m *Manager) Contributions() []extension.Contribution { |
| 267 | var out []extension.Contribution |
| 268 | for _, client := range m.Clients() { |
| 269 | rt := client.rt |
| 270 | source := extension.ContributionSource{ |
| 271 | Scope: extension.ScopePlugin, |
| 272 | PluginID: client.pluginID, |
| 273 | Version: client.version, |
| 274 | Origin: "extension-runtime", |
| 275 | } |
| 276 | for _, point := range rt.Intercepts { |
| 277 | if !kernelID(point) { |
| 278 | slog.Debug("sidecar: skipping interceptor point outside the kernel ID contract", "plugin", client.pluginID, "point", point) |
| 279 | continue |
| 280 | } |
| 281 | out = append(out, extension.Contribution{ |
| 282 | Kind: extension.KindInterceptor, |
| 283 | ID: point, |
| 284 | Source: source, |
| 285 | Priority: rt.Priority, |
| 286 | Payload: InterceptorDecl{PluginID: client.pluginID, Point: point, Priority: rt.Priority}, |
| 287 | }) |
| 288 | } |
| 289 | for _, slot := range rt.Replaces { |
| 290 | if !kernelID(slot) { |
| 291 | slog.Debug("sidecar: skipping replacement slot outside the kernel ID contract", "plugin", client.pluginID, "slot", slot) |
| 292 | continue |
| 293 | } |
| 294 | out = append(out, extension.Contribution{ |
| 295 | Kind: extension.KindStrategy, |
| 296 | ID: slot, |
| 297 | Source: source, |
| 298 | Priority: rt.Priority, |
| 299 | Payload: StrategyDecl{PluginID: client.pluginID, Slots: []extension.Slot{extension.Slot(slot)}}, |
| 300 | }) |
| 301 | } |
| 302 | result := client.Handshake() |
| 303 | prefix := "plugin/" + client.pluginID + "/" |
| 304 | for _, desc := range result.Providers { |
| 305 | ref := strings.TrimPrefix(desc.Ref, prefix) |
| 306 | if !extension.IsProviderRef(ref) { |
| 307 | slog.Debug("sidecar: skipping provider ref outside the kernel ID contract", "plugin", client.pluginID, "ref", desc.Ref) |
| 308 | continue |
| 309 | } |
| 310 | out = append(out, extension.Contribution{ |
| 311 | Kind: extension.KindProvider, |
| 312 | ID: ref, |
| 313 | Source: source, |
| 314 | Payload: ProviderDecl{PluginID: client.pluginID, Descriptor: desc}, |
| 315 | }) |
| 316 | } |
| 317 | for _, decl := range result.UIActions { |
| 318 | if !kernelID(decl.ActionID) { |
| 319 | slog.Debug("sidecar: skipping UI action outside the kernel ID contract", "plugin", client.pluginID, "action", decl.ActionID) |
| 320 | continue |
| 321 | } |
| 322 | out = append(out, extension.Contribution{ |
| 323 | Kind: extension.KindUIAction, |
| 324 | ID: decl.ActionID, |
| 325 | Source: source, |
| 326 | Payload: UIActionDecl{PluginID: client.pluginID, Decl: decl}, |
| 327 | }) |
| 328 | } |
| 329 | } |
| 330 | return out |
| 331 | } |
| 332 | |
| 333 | // kernelID mirrors the kernel's generic ID hygiene (non-empty, no |
| 334 | // whitespace); boot's legacy assembly uses the same rule. |
| 335 | func kernelID(id string) bool { |
| 336 | id = strings.TrimSpace(id) |
| 337 | return id != "" && !strings.ContainsAny(id, " \t\n") |
| 338 | } |
| 339 |