返回 DeepSeek-Reasonix
plan.go
1 package sidecar
2
3 import (
4 "context"
5 "fmt"
6 "log/slog"
7 "strings"
8
9 "reasonix/internal/extension"
10 "reasonix/internal/extension/protocol"
11 "reasonix/internal/pluginpkg"
12 "reasonix/internal/secrets"
13 )
14
15 // PluginComponentID returns the dependency-graph component ID for an installed
16 // native runtime package.
17 func PluginComponentID(pluginName string) extension.ComponentID {
18 return extension.ComponentID("plugin/" + strings.TrimSpace(pluginName))
19 }
20
21 // PluginNameFromComponentID extracts the plugin package name from a
22 // plugin/<name> component ID. Non-plugin IDs return "".
23 func PluginNameFromComponentID(id extension.ComponentID) string {
24 const prefix = "plugin/"
25 s := string(id)
26 if !strings.HasPrefix(s, prefix) {
27 return ""
28 }
29 return strings.TrimSpace(s[len(prefix):])
30 }
31
32 // StartPackagesWithPlan starts Added/Reloaded packages and adopts Unchanged
33 // clients from previous. previous may be nil. Required start failures fail
34 // the whole call and close the new Manager's resources.
35 func StartPackagesWithPlan(ctx context.Context, home string, sessionCtx protocol.SessionContext, ui UIHandler, previous *Manager, plan *extension.RuntimePlan) (*Manager, []string, error) {
36 packages, warnings := LoadRuntimePackages(home)
37 if plan == nil || plan.IsNoOp() && previous == nil {
38 startupCtx, cancel := context.WithTimeout(ctx, packageStartupBudget)
39 defer cancel()
40 m, runtimeWarnings, err := startLoadedPackages(startupCtx, packages, sessionCtx, ui, StartClient)
41 warnings = append(warnings, runtimeWarnings...)
42 return m, warnings, err
43 }
44 if plan.IsNoOp() && previous != nil && !plan.RestartUnchangedSidecars {
45 // No component change: adopt every live client from previous.
46 return adoptAll(previous), warnings, nil
47 }
48
49 activate := map[string]bool{}
50 for _, id := range plan.Added {
51 if name := PluginNameFromComponentID(id); name != "" {
52 activate[name] = true
53 }
54 }
55 for _, id := range plan.Reloaded {
56 if name := PluginNameFromComponentID(id); name != "" {
57 activate[name] = true
58 }
59 }
60 unchanged := map[string]bool{}
61 for _, id := range plan.Unchanged {
62 if name := PluginNameFromComponentID(id); name != "" {
63 if plan.RestartUnchangedSidecars {
64 activate[name] = true
65 } else {
66 unchanged[name] = true
67 }
68 }
69 }
70
71 var toStart []pluginpkg.InstalledPackage
72 for _, item := range packages {
73 name := item.Installed.Name
74 if activate[name] {
75 toStart = append(toStart, item)
76 }
77 }
78
79 startupCtx, cancel := context.WithTimeout(ctx, packageStartupBudget)
80 defer cancel()
81 m, runtimeWarnings, err := startLoadedPackages(startupCtx, toStart, sessionCtx, ui, StartClient)
82 warnings = append(warnings, runtimeWarnings...)
83 if n := len(toStart); n > 0 {
84 extension.DefaultLifecycleMetrics.SidecarStarts.Add(uint64(n))
85 }
86 if err != nil {
87 return m, warnings, err
88 }
89
90 // Adopt unchanged clients from previous. Detach so previous.Close after
91 // publish does not kill still-active packages. Track detaches so a later
92 // activation failure can reattach ONLY these unchanged clients.
93 m.planAdopted = map[string]*Client{}
94 rollback := func() {
95 m.RollbackPlanStart(previous)
96 }
97 if previous != nil {
98 for name := range unchanged {
99 client := previous.Detach(name)
100 if client == nil {
101 // Unchanged in the graph but no live client: start it now.
102 for _, item := range packages {
103 if item.Installed.Name != name {
104 continue
105 }
106 fresh, startErr := startOne(startupCtx, item, sessionCtx, ui)
107 if startErr != nil {
108 if item.Package.Manifest.Runtime != nil && item.Package.Manifest.Runtime.Required {
109 rollback()
110 return nil, warnings, &RequiredStartError{Plugin: name, Err: startErr}
111 }
112 warnings = append(warnings, fmt.Sprintf("%s: optional extension runtime failed to start: %v", name, startErr))
113 break
114 }
115 if adoptErr := m.Adopt(name, fresh); adoptErr != nil {
116 _ = fresh.Close()
117 rollback()
118 return nil, warnings, adoptErr
119 }
120 break
121 }
122 continue
123 }
124 if adoptErr := m.Adopt(name, client); adoptErr != nil {
125 if reattachErr := previous.Adopt(name, client); reattachErr != nil {
126 _ = client.Close()
127 }
128 rollback()
129 return nil, warnings, adoptErr
130 }
131 m.planAdopted[name] = client
132 extension.DefaultLifecycleMetrics.SidecarAdopts.Add(1)
133 }
134 }
135 return m, warnings, nil
136 }
137
138 func adoptAll(previous *Manager) *Manager {
139 m := &Manager{clients: make(map[string]*Client)}
140 if previous == nil {
141 return m
142 }
143 for _, client := range previous.Clients() {
144 id := client.PluginID()
145 if c := previous.Detach(id); c != nil {
146 m.clients[id] = c
147 }
148 }
149 return m
150 }
151
152 func startOne(ctx context.Context, item pluginpkg.InstalledPackage, sessionCtx protocol.SessionContext, ui UIHandler) (*Client, error) {
153 pluginID := item.Installed.Name
154 var binder UIBinder
155 if b, ok := ui.(UIBinder); ok {
156 binder = b
157 }
158 clientUI := ui
159 if binder != nil {
160 clientUI = binder.HandlerFor(pluginID)
161 }
162 return StartClient(ctx, ClientOptions{
163 Package: item.Package,
164 Installed: item.Installed,
165 Session: sessionCtx,
166 UI: clientUI,
167 OnCrash: func(err error) {
168 slog.Warn("extension sidecar crashed", "plugin", pluginID, "err", secrets.RedactError(err))
169 if binder != nil {
170 binder.ClientCrashed(pluginID)
171 }
172 },
173 })
174 }
175
176 // Detach removes a client from the manager without closing it. Returns nil
177 // when the plugin is not present or the manager is closed.
178 func (m *Manager) Detach(pluginID string) *Client {
179 if m == nil {
180 return nil
181 }
182 m.mu.Lock()
183 defer m.mu.Unlock()
184 if m.closed {
185 return nil
186 }
187 client := m.clients[pluginID]
188 delete(m.clients, pluginID)
189 return client
190 }
191
192 // Adopt takes ownership of an already-running client. Fails if the manager is
193 // closed or the plugin ID is already registered.
194 func (m *Manager) Adopt(pluginID string, client *Client) error {
195 if m == nil {
196 return fmt.Errorf("sidecar: nil Manager")
197 }
198 if client == nil {
199 return fmt.Errorf("sidecar: nil Client")
200 }
201 pluginID = strings.TrimSpace(pluginID)
202 if pluginID == "" {
203 return fmt.Errorf("sidecar: empty plugin id")
204 }
205 m.mu.Lock()
206 defer m.mu.Unlock()
207 if m.closed {
208 return fmt.Errorf("sidecar: manager closed")
209 }
210 if m.clients == nil {
211 m.clients = make(map[string]*Client)
212 }
213 if _, exists := m.clients[pluginID]; exists {
214 return fmt.Errorf("sidecar: plugin %q already registered", pluginID)
215 }
216 m.clients[pluginID] = client
217 return nil
218 }
219
220 // Drain closes the listed plugin clients (if present) and removes them. Other
221 // clients are left running. Used after publish to retire Removed/Reloaded
222 // packages still held by the old generation's manager.
223 func (m *Manager) Drain(pluginIDs ...string) {
224 if m == nil {
225 return
226 }
227 for _, id := range pluginIDs {
228 if c := m.Detach(id); c != nil {
229 _ = c.Close()
230 }
231 }
232 }
233
234 // RollbackPlanStart undoes StartPackagesWithPlan ownership transfer:
235 // - Unchanged clients recorded in planAdopted are reattached to previous
236 // - Remaining clients (Added/Reloaded fresh starts) are closed via m.Close()
237 //
238 // Adopt errors are not swallowed: failed reattach closes the client to avoid leak.
239 func (m *Manager) RollbackPlanStart(previous *Manager) {
240 if m == nil {
241 return
242 }
243 m.mu.Lock()
244 adopted := m.planAdopted
245 m.planAdopted = nil
246 m.mu.Unlock()
247 for name, client := range adopted {
248 if c := m.Detach(name); c != nil {
249 client = c
250 }
251 if previous == nil {
252 _ = client.Close()
253 continue
254 }
255 if err := previous.Adopt(name, client); err != nil {
256 // Previous still holds a Reloaded old client under the same name, or
257 // is closed — close the orphaned client so the process does not leak.
258 _ = client.Close()
259 }
260 }
261 _ = m.Close()
262 }
263
264 // DrainPlan closes clients matching the plan's Removed and Reloaded sets.
265 func (m *Manager) DrainPlan(plan *extension.RuntimePlan) {
266 if m == nil || plan == nil {
267 return
268 }
269 var ids []string
270 for _, id := range plan.Removed {
271 if name := PluginNameFromComponentID(id); name != "" {
272 ids = append(ids, name)
273 }
274 }
275 for _, id := range plan.Reloaded {
276 if name := PluginNameFromComponentID(id); name != "" {
277 ids = append(ids, name)
278 }
279 }
280 m.Drain(ids...)
281 }
282
282 lines GO