diff --git a/apps/backend/internal/plugins/delivery/deliverer.go b/apps/backend/internal/plugins/delivery/deliverer.go index dc1047df94..0817074786 100644 --- a/apps/backend/internal/plugins/delivery/deliverer.go +++ b/apps/backend/internal/plugins/delivery/deliverer.go @@ -38,10 +38,11 @@ import ( // Tests override these via Option so retry/backoff loops don't rely on real // sleeps. const ( - DefaultRequestTimeout = 10 * time.Second - DefaultQueueSize = 100 - DefaultRingBufferCap = 100 - DefaultRingBufferTTL = 5 * time.Minute + DefaultRequestTimeout = 10 * time.Second + DefaultQueueSize = 100 + DefaultRingBufferCap = 100 + DefaultRingBufferTTL = 5 * time.Minute + DefaultOverflowLogInterval = time.Minute ) // DefaultRetryDelays is the backoff schedule between delivery attempts @@ -115,6 +116,13 @@ func WithNow(fn func() time.Time) Option { return func(d *Deliverer) { d.nowFn = fn } } +// WithOverflowLogInterval overrides the per-plugin ring-buffer overflow log +// aggregation window. Production uses one minute; tests inject a shorter +// window or advance an injected clock instead of sleeping. +func WithOverflowLogInterval(interval time.Duration) Option { + return func(d *Deliverer) { d.overflowLogInterval = interval } +} + // withOnProcessed is an unexported Option (only constructible from within // this package) used by tests to observe exactly when a worker finishes // handling each queued Delivery, so a test can wait for "all N events have @@ -136,12 +144,13 @@ type Deliverer struct { lister PluginLister log *logger.Logger - requestTimeout time.Duration - retryDelays []time.Duration - queueSize int - ringBufferCap int - ringBufferTTL time.Duration - nowFn func() time.Time + requestTimeout time.Duration + retryDelays []time.Duration + queueSize int + ringBufferCap int + ringBufferTTL time.Duration + overflowLogInterval time.Duration + nowFn func() time.Time mu sync.Mutex workers map[string]*pluginWorker @@ -160,17 +169,18 @@ type Deliverer struct { // from the current plugin set. func New(eventBus bus.EventBus, transport Transport, lister PluginLister, log *logger.Logger, opts ...Option) *Deliverer { d := &Deliverer{ - eventBus: eventBus, - transport: transport, - lister: lister, - log: log, - requestTimeout: DefaultRequestTimeout, - retryDelays: DefaultRetryDelays(), - queueSize: DefaultQueueSize, - ringBufferCap: DefaultRingBufferCap, - ringBufferTTL: DefaultRingBufferTTL, - nowFn: time.Now, - workers: make(map[string]*pluginWorker), + eventBus: eventBus, + transport: transport, + lister: lister, + log: log, + requestTimeout: DefaultRequestTimeout, + retryDelays: DefaultRetryDelays(), + queueSize: DefaultQueueSize, + ringBufferCap: DefaultRingBufferCap, + ringBufferTTL: DefaultRingBufferTTL, + overflowLogInterval: DefaultOverflowLogInterval, + nowFn: time.Now, + workers: make(map[string]*pluginWorker), } for _, opt := range opts { opt(d) @@ -267,11 +277,13 @@ func (d *Deliverer) newWorker(id string) *pluginWorker { return &pluginWorker{ id: id, deps: &workerDeps{ - transport: d.transport, - requestTimeout: d.requestTimeout, - retryDelays: d.retryDelays, - log: d.log, - onProcessed: d.onProcessed, + transport: d.transport, + requestTimeout: d.requestTimeout, + retryDelays: d.retryDelays, + log: d.log, + nowFn: d.nowFn, + overflowLogInterval: d.overflowLogInterval, + onProcessed: d.onProcessed, }, queue: make(chan Delivery, d.queueSize), buffer: newRingBuffer(d.ringBufferCap, d.ringBufferTTL, d.nowFn), @@ -357,11 +369,13 @@ func (d *Deliverer) makeHandler(w *pluginWorker, pattern string) bus.EventHandle // delivery, factored out of Deliverer so pluginWorker doesn't hold a back // reference to it. type workerDeps struct { - transport Transport - requestTimeout time.Duration - retryDelays []time.Duration - log *logger.Logger - onProcessed func(pluginID string, d Delivery) + transport Transport + requestTimeout time.Duration + retryDelays []time.Duration + log *logger.Logger + nowFn func() time.Time + overflowLogInterval time.Duration + onProcessed func(pluginID string, d Delivery) } // pluginWorker owns sequential delivery for one plugin: a bounded queue @@ -388,9 +402,12 @@ type pluginWorker struct { deliverCtx context.Context cancelDeliver context.CancelFunc - mu sync.Mutex - subs []bus.Subscription - status string + mu sync.Mutex + subs []bus.Subscription + status string + overflowLogAt time.Time + overflowLogSet bool + overflowDropped int } // start launches the worker's run loop. Must be called at most once. @@ -479,8 +496,7 @@ func (w *pluginWorker) run() { func (w *pluginWorker) process(d Delivery) { if w.snapshotStatus() == store.StatusError { if dropped := w.buffer.Add(d); dropped != "" { - w.deps.log.Warn("plugin event ring buffer overflow, dropped oldest buffered event", - zap.String("plugin_id", w.id), zap.String("dropped_delivery_id", dropped)) + w.logOverflow(dropped) } } else { w.deliverWithRetry(w.deliverCtx, d) @@ -490,6 +506,30 @@ func (w *pluginWorker) process(d Delivery) { } } +// logOverflow rate-limits ring-buffer overflow warnings per plugin while +// retaining an aggregate count of suppressed drops. The first drop is logged +// immediately; the first drop after the interval logs every drop accumulated +// since the previous warning. +func (w *pluginWorker) logOverflow(dropped string) { + now := w.deps.nowFn().UTC() + w.overflowDropped++ + count := w.overflowDropped + shouldLog := !w.overflowLogSet || !now.Before(w.overflowLogAt.Add(w.deps.overflowLogInterval)) + if shouldLog { + w.overflowDropped = 0 + w.overflowLogAt = now + w.overflowLogSet = true + } + + if !shouldLog { + return + } + w.deps.log.Warn("plugin event ring buffer overflow, dropped oldest buffered event", + zap.String("plugin_id", w.id), + zap.String("latest_dropped_delivery_id", dropped), + zap.Int("dropped_count", count)) +} + // deliverWithRetry attempts d up to 1+len(retryDelays) times, sleeping // (interruptibly) between attempts per the configured backoff schedule. // Logs and gives up (event is lost from the live-delivery path; a diff --git a/apps/backend/internal/plugins/delivery/deliverer_test.go b/apps/backend/internal/plugins/delivery/deliverer_test.go index 1dd753288b..e27fafea1d 100644 --- a/apps/backend/internal/plugins/delivery/deliverer_test.go +++ b/apps/backend/internal/plugins/delivery/deliverer_test.go @@ -12,6 +12,9 @@ import ( "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/plugins/store" "github.com/kandev/kandev/pkg/pluginsdk" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" ) // fakeTransport is a controllable Transport for tests: DeliverEvent calls @@ -351,6 +354,124 @@ func TestDeliverer_RingBufferOverflowDropsOldest(t *testing.T) { requireTimeout(t, arrivedCh, 150*time.Millisecond, "a 3rd replayed event") } +func TestDeliverer_RingBufferOverflowLogsAggregatedDrops(t *testing.T) { + clock := time.Date(2026, time.August, 2, 12, 0, 0, 0, time.UTC) + var clockMu sync.Mutex + now := func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return clock + } + core, observed := observer.New(zapcore.WarnLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap(): %v", err) + } + + eventBus := bus.NewMemoryEventBus(logger.Default()) + lister := &fakeLister{} + lister.set(PluginRecord{ID: "plug1", EventSubjects: []string{"task.*"}, Status: store.StatusError}) + processedCh := make(chan struct{}, 10) + d := New(eventBus, &fakeTransport{}, lister, log, + WithRetryDelays(nil), + WithRingBuffer(1, 5*time.Minute), + WithNow(now), + WithOverflowLogInterval(time.Minute), + withOnProcessed(func(string, Delivery) { processedCh <- struct{}{} })) + t.Cleanup(d.Stop) + d.Refresh() + + ctx := context.Background() + for _, subject := range []string{"task.created", "task.updated", "task.deleted"} { + if err := eventBus.Publish(ctx, subject, bus.NewEvent(subject, "test", map[string]interface{}{})); err != nil { + t.Fatalf("Publish(%s): %v", subject, err) + } + } + for i := 0; i < 3; i++ { + requireNoTimeout(t, processedCh, 2*time.Second, fmt.Sprintf("event %d buffered", i+1)) + } + + entries := observed.All() + if len(entries) != 1 { + t.Fatalf("overflow warnings after two drops = %d, want 1", len(entries)) + } + if got := entries[0].ContextMap()["dropped_count"]; got != int64(1) { + t.Fatalf("first dropped_count = %v, want 1", got) + } + if got := entries[0].ContextMap()["latest_dropped_delivery_id"]; got == nil || got == "" { + t.Fatalf("first latest_dropped_delivery_id = %v, want a non-empty id", got) + } + if _, ok := entries[0].ContextMap()["dropped_delivery_id"]; ok { + t.Fatal("first overflow warning still uses the ambiguous dropped_delivery_id field") + } + + clockMu.Lock() + clock = clock.Add(time.Minute) + clockMu.Unlock() + if err := eventBus.Publish(ctx, "task.archived", bus.NewEvent("task.archived", "test", map[string]interface{}{})); err != nil { + t.Fatalf("Publish(task.archived): %v", err) + } + requireNoTimeout(t, processedCh, 2*time.Second, "post-interval event buffered") + + entries = observed.All() + if len(entries) != 2 { + t.Fatalf("overflow warnings after interval = %d, want 2", len(entries)) + } + if got := entries[1].ContextMap()["dropped_count"]; got != int64(2) { + t.Fatalf("aggregated dropped_count = %v, want 2", got) + } + if got := entries[1].ContextMap()["latest_dropped_delivery_id"]; got == nil || got == "" { + t.Fatalf("aggregated latest_dropped_delivery_id = %v, want a non-empty id", got) + } + if _, ok := entries[1].ContextMap()["dropped_delivery_id"]; ok { + t.Fatal("aggregated overflow warning still uses the ambiguous dropped_delivery_id field") + } +} + +func TestDeliverer_RingBufferOverflowLimiterIsPerPlugin(t *testing.T) { + core, observed := observer.New(zapcore.WarnLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap(): %v", err) + } + + eventBus := bus.NewMemoryEventBus(logger.Default()) + lister := &fakeLister{} + lister.set( + PluginRecord{ID: "plug1", EventSubjects: []string{"task.*"}, Status: store.StatusError}, + PluginRecord{ID: "plug2", EventSubjects: []string{"task.*"}, Status: store.StatusError}, + ) + processedCh := make(chan struct{}, 10) + d := New(eventBus, &fakeTransport{}, lister, log, + WithRetryDelays(nil), + WithRingBuffer(1, 5*time.Minute), + WithOverflowLogInterval(time.Minute), + withOnProcessed(func(string, Delivery) { processedCh <- struct{}{} })) + t.Cleanup(d.Stop) + d.Refresh() + + for _, subject := range []string{"task.created", "task.updated"} { + if err := eventBus.Publish(context.Background(), subject, bus.NewEvent(subject, "test", map[string]interface{}{})); err != nil { + t.Fatalf("Publish(%s): %v", subject, err) + } + } + for i := 0; i < 4; i++ { + requireNoTimeout(t, processedCh, 2*time.Second, fmt.Sprintf("plugin event %d buffered", i+1)) + } + + entries := observed.All() + if len(entries) != 2 { + t.Fatalf("overflow warnings for two plugins = %d, want 2", len(entries)) + } + seen := map[string]bool{} + for _, entry := range entries { + seen[entry.ContextMap()["plugin_id"].(string)] = true + } + if !seen["plug1"] || !seen["plug2"] { + t.Fatalf("overflow warning plugin ids = %v, want both plugins", seen) + } +} + func TestDeliverer_RefreshStopsDeliveryWhenPluginRemoved(t *testing.T) { arrivedCh := make(chan struct{}, 5) transport := &fakeTransport{} diff --git a/apps/backend/internal/plugins/diagnostics.go b/apps/backend/internal/plugins/diagnostics.go new file mode 100644 index 0000000000..5596a0c634 --- /dev/null +++ b/apps/backend/internal/plugins/diagnostics.go @@ -0,0 +1,73 @@ +package plugins + +import ( + "os" + "regexp" + "strings" + "sync" + "unicode/utf8" +) + +// maxPluginErrorBytes bounds the diagnostic persisted into a plugin record. +// Failure messages can originate in subprocess/transport errors, so keeping +// them bounded prevents an unexpectedly verbose error from bloating the +// record or the settings API response. +const maxPluginErrorBytes = 2048 + +var ( + pluginBearerTokenPattern = regexp.MustCompile(`(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+`) + pluginLabeledSecretPattern = regexp.MustCompile(`(?i)\b((?:[A-Za-z0-9]+[_-])*(?:password|passwd|passphrase|secret|token|api[_ -]?(?:key|token|secret)|access[_ -]?token|refresh[_ -]?token|pat))\b(\s*[:=]\s*)(?:"[^"]*"|'[^']*'|[^\s,;}]+)`) + pluginPATPattern = regexp.MustCompile(`\b(?:github_pat_[A-Za-z0-9_]+|gh[pousr]_[A-Za-z0-9]+|glpat-[A-Za-z0-9_-]+)\b`) + pluginHomePathPattern = regexp.MustCompile(`(/Users/[^/\s]+|/home/[^/\s]+|/root)(/|$)`) + pluginWindowsHomePattern = regexp.MustCompile(`(?i)([A-Za-z]:[/\\]Users[/\\][^/\\\s]+)([/\\]|$)`) + pluginURLCredentialPattern = regexp.MustCompile(`(?i)(https?|ftp)://[^:@/\s]+:[^@/\s]+@`) + + pluginHomePatternOnce sync.Once + cachedPluginHomePattern *regexp.Regexp +) + +// normalizePluginError turns a runtime failure into a compact, single-line +// diagnostic suitable for persistence and display. Runtime errors can include +// arbitrary go-plugin subprocess output, so credential-like values and home +// paths are removed before the bounded message is stored. +func normalizePluginError(err error) string { + if err == nil { + return "" + } + message := strings.Join(strings.Fields(err.Error()), " ") + message = strings.ToValidUTF8(message, "�") + message = redactPluginError(message) + if len([]byte(message)) <= maxPluginErrorBytes { + return message + } + + return truncatePluginError(message) +} + +func truncatePluginError(message string) string { + const marker = "…" + content := []byte(message) + cut := maxPluginErrorBytes - len([]byte(marker)) + for cut > 0 && cut < len(content) && !utf8.RuneStart(content[cut]) { + cut-- + } + return string(content[:cut]) + marker +} + +func redactPluginError(message string) string { + pluginHomePatternOnce.Do(func() { + if home, err := os.UserHomeDir(); err == nil && home != "" && home != "/" { + cachedPluginHomePattern = regexp.MustCompile(regexp.QuoteMeta(home) + `([/\\]|$)`) + } + }) + if cachedPluginHomePattern != nil { + message = cachedPluginHomePattern.ReplaceAllString(message, "~$1") + } + message = pluginHomePathPattern.ReplaceAllString(message, "~$2") + message = pluginWindowsHomePattern.ReplaceAllString(message, "~$2") + message = pluginURLCredentialPattern.ReplaceAllString(message, "${1}://[REDACTED]@") + message = pluginBearerTokenPattern.ReplaceAllString(message, "Bearer [REDACTED]") + message = pluginLabeledSecretPattern.ReplaceAllString(message, "${1}${2}[REDACTED]") + message = pluginPATPattern.ReplaceAllString(message, "[REDACTED]") + return message +} diff --git a/apps/backend/internal/plugins/diagnostics_test.go b/apps/backend/internal/plugins/diagnostics_test.go new file mode 100644 index 0000000000..665eb1ec89 --- /dev/null +++ b/apps/backend/internal/plugins/diagnostics_test.go @@ -0,0 +1,113 @@ +package plugins + +import ( + "errors" + "os" + "strings" + "testing" +) + +func TestNormalizePluginErrorCollapsesWhitespace(t *testing.T) { + got := normalizePluginError(errors.New(" handshake\nfailed\twhile starting ")) + if got != "handshake failed while starting" { + t.Fatalf("normalizePluginError() = %q, want %q", got, "handshake failed while starting") + } +} + +func TestNormalizePluginErrorBoundsUTF8Message(t *testing.T) { + got := normalizePluginError(errors.New(strings.Repeat("é", maxPluginErrorBytes))) + if len([]byte(got)) > maxPluginErrorBytes { + t.Fatalf("normalizePluginError() returned %d bytes, want <= %d", len([]byte(got)), maxPluginErrorBytes) + } + if !strings.HasSuffix(got, "…") { + t.Fatalf("normalizePluginError() = %q, want a truncation marker", got) + } +} + +func TestNormalizePluginErrorNilIsEmpty(t *testing.T) { + if got := normalizePluginError(nil); got != "" { + t.Fatalf("normalizePluginError(nil) = %q, want empty", got) + } +} + +func TestNormalizePluginErrorRedactsCredentialsAndHomePath(t *testing.T) { + home, err := os.UserHomeDir() + if err != nil { + t.Fatalf("os.UserHomeDir() error: %v", err) + } + + const ( + classicPAT = "ghp_abcdefghijklmnopqrstuvwxyz1234567890AB" + bearer = "eyJhbGciOiJIUzI1NiJ9.super-secret-signature" + password = "hunter2-rocks" + token = "plugin-token-value" + secret = "s3cret-value" + apiKey = "key-value-123" + apiToken = "api-token-value" + clientSecret = "client-secret-value" + urlPassword = "url-password-value" + ) + cases := []struct { + name string + input string + secret []string + want string + }{ + { + name: "personal access token", + input: "plugin stdout included PAT " + classicPAT, + secret: []string{classicPAT}, + want: "[REDACTED]", + }, + { + name: "bearer authorization", + input: "connect failed: Authorization: Bearer " + bearer, + secret: []string{bearer}, + want: "Bearer [REDACTED]", + }, + { + name: "labeled secrets", + input: "password=" + password + " token: " + token + " secret='" + secret + "' api_key=\"" + apiKey + "\" api_token=" + apiToken, + secret: []string{password, token, secret, apiKey, apiToken}, + want: "[REDACTED]", + }, + { + name: "compound labeled secret", + input: "plugin failed with client_secret=" + clientSecret, + secret: []string{clientSecret}, + want: "client_secret=[REDACTED]", + }, + { + name: "URL credentials", + input: "dial failed: https://admin:" + urlPassword + "@internal.example/api", + secret: []string{urlPassword}, + want: "https://[REDACTED]@internal.example/api", + }, + { + name: "home path", + input: home + "/.kandev/plugins/acme/server: handshake failed", + secret: []string{home + "/"}, + want: "~/.kandev/plugins/acme/server", + }, + { + name: "windows home path", + input: `plugin failed at C:\Users\alice\.kandev\plugins\acme\server: handshake failed`, + secret: []string{`C:\Users\alice\`}, + want: `~\.kandev\plugins\acme\server`, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := normalizePluginError(errors.New(tc.input)) + for _, value := range tc.secret { + if strings.Contains(got, value) { + t.Fatalf("normalizePluginError() leaked %q in %q", value, got) + } + } + if !strings.Contains(got, tc.want) { + t.Fatalf("normalizePluginError() = %q, want it to contain %q", got, tc.want) + } + }) + } +} diff --git a/apps/backend/internal/plugins/registry.go b/apps/backend/internal/plugins/registry.go index 2a7396af9a..189da5d425 100644 --- a/apps/backend/internal/plugins/registry.go +++ b/apps/backend/internal/plugins/registry.go @@ -4,6 +4,7 @@ import ( "fmt" "sort" "sync" + "time" "github.com/kandev/kandev/internal/plugins/store" ) @@ -15,7 +16,8 @@ import ( // // Get and List return copies of the stored *store.Record so callers cannot // mutate registry state by holding onto a returned pointer; all writes go -// through Add / Remove / SetStatus / SetRestartCount. +// through Add / Remove / SetStatus / SetRuntimeState / SetAutoUpdate / +// SetRestartCount. type Registry struct { mu sync.RWMutex byID map[string]*store.Record @@ -37,7 +39,7 @@ func (r *Registry) Load(s store.Store) error { byID := make(map[string]*store.Record, len(records)) for _, rec := range records { - byID[rec.ID] = rec + byID[rec.ID] = cloneRecord(rec) } r.mu.Lock() @@ -54,8 +56,7 @@ func (r *Registry) Get(id string) (*store.Record, bool) { if !ok { return nil, false } - clone := *rec - return &clone, true + return cloneRecord(rec), true } // List returns a copy of every registered record, sorted by id for @@ -65,8 +66,7 @@ func (r *Registry) List() []*store.Record { defer r.mu.RUnlock() out := make([]*store.Record, 0, len(r.byID)) for _, rec := range r.byID { - clone := *rec - out = append(out, &clone) + out = append(out, cloneRecord(rec)) } sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) return out @@ -76,8 +76,7 @@ func (r *Registry) List() []*store.Record { func (r *Registry) Add(record *store.Record) { r.mu.Lock() defer r.mu.Unlock() - clone := *record - r.byID[record.ID] = &clone + r.byID[record.ID] = cloneRecord(record) } // Remove deletes the record for id, if present. A no-op if id is unknown. @@ -99,8 +98,7 @@ func (r *Registry) SetStatus(id string, status Status) (*store.Record, bool) { return nil, false } rec.Status = status - clone := *rec - return &clone, true + return cloneRecord(rec), true } // SetAutoUpdate updates the in-memory per-plugin auto-update override for id @@ -115,8 +113,7 @@ func (r *Registry) SetAutoUpdate(id string, v *bool) (*store.Record, bool) { return nil, false } rec.AutoUpdate = v - clone := *rec - return &clone, true + return cloneRecord(rec), true } // SetRestartCount updates the in-memory restart count for id and returns a @@ -131,6 +128,40 @@ func (r *Registry) SetRestartCount(id string, count int) (*store.Record, bool) { return nil, false } rec.RestartCount = count - clone := *rec - return &clone, true + return cloneRecord(rec), true +} + +// SetRuntimeState updates the lifecycle status and persisted failure +// diagnostic for id and returns a copy of the resulting record. The caller is +// responsible for FSM validation and persistence; this method only performs +// the in-memory mutation under the registry lock. +func (r *Registry) SetRuntimeState(id string, status Status, lastError string, lastErrorAt *time.Time) (*store.Record, bool) { + r.mu.Lock() + defer r.mu.Unlock() + rec, ok := r.byID[id] + if !ok { + return nil, false + } + rec.Status = status + rec.LastError = lastError + if lastErrorAt == nil { + rec.LastErrorAt = nil + } else { + at := *lastErrorAt + rec.LastErrorAt = &at + } + return cloneRecord(rec), true +} + +func cloneRecord(rec *store.Record) *store.Record { + clone := *rec + if rec.AutoUpdate != nil { + autoUpdate := *rec.AutoUpdate + clone.AutoUpdate = &autoUpdate + } + if rec.LastErrorAt != nil { + at := *rec.LastErrorAt + clone.LastErrorAt = &at + } + return &clone } diff --git a/apps/backend/internal/plugins/registry_test.go b/apps/backend/internal/plugins/registry_test.go index 45ac09e2cf..980106c912 100644 --- a/apps/backend/internal/plugins/registry_test.go +++ b/apps/backend/internal/plugins/registry_test.go @@ -105,13 +105,19 @@ func TestRegistrySetStatusMissingReturnsNotOK(t *testing.T) { func TestRegistryGetReturnsIndependentCopy(t *testing.T) { reg := NewRegistry() - reg.Add(&store.Record{Manifest: *testManifest("kandev-plugin-slack"), Status: store.StatusRegistered}) + autoUpdate := true + reg.Add(&store.Record{ + Manifest: *testManifest("kandev-plugin-slack"), + Status: store.StatusRegistered, + AutoUpdate: &autoUpdate, + }) rec, ok := reg.Get("kandev-plugin-slack") if !ok { t.Fatalf("Get() expected ok = true") } rec.Status = store.StatusActive // mutate the returned copy + *rec.AutoUpdate = false // mutate the returned pointer copy fresh, ok := reg.Get("kandev-plugin-slack") if !ok { @@ -120,6 +126,9 @@ func TestRegistryGetReturnsIndependentCopy(t *testing.T) { if fresh.Status != store.StatusRegistered { t.Fatalf("mutating a Get() result leaked into the registry: Status = %q, want %q", fresh.Status, store.StatusRegistered) } + if fresh.AutoUpdate == nil || !*fresh.AutoUpdate { + t.Fatalf("mutating a Get() result leaked into the registry: AutoUpdate = %v, want true", fresh.AutoUpdate) + } } func TestRegistrySetRestartCountUpdatesRecord(t *testing.T) { diff --git a/apps/backend/internal/plugins/runtime/manager.go b/apps/backend/internal/plugins/runtime/manager.go index d05f060e33..9d451f34f5 100644 --- a/apps/backend/internal/plugins/runtime/manager.go +++ b/apps/backend/internal/plugins/runtime/manager.go @@ -88,7 +88,7 @@ func WithStartTimeout(d time.Duration) Option { type Manager struct { pluginsDir string log *logger.Logger - onStatusChange func(id string, healthy bool) + onStatusChange func(id string, healthy bool, reason error) pingInterval time.Duration maxConsecutiveFailures int @@ -124,9 +124,11 @@ type Manager struct { // store.FSStore persists records under — KANDEV_PLUGIN_DATA_DIR for plugin // id "x" is pluginsDir/x/data). onStatusChange is invoked from the // supervision loop's own goroutine whenever a running plugin's health -// transitions (degraded -> unhealthy, or a restart recovers it); it must -// not block. May be nil (e.g. tests that don't care about transitions). -func NewManager(pluginsDir string, onStatusChange func(id string, healthy bool), log *logger.Logger, opts ...Option) *Manager { +// transitions (degraded -> unhealthy, or a restart recovers it); reason is +// the triggering ping/process/restart error for unhealthy transitions and nil +// on recovery. The callback must not block. May be nil (e.g. tests that don't +// care about transitions). +func NewManager(pluginsDir string, onStatusChange func(id string, healthy bool, reason error), log *logger.Logger, opts ...Option) *Manager { m := &Manager{ pluginsDir: pluginsDir, log: log, @@ -236,27 +238,34 @@ func (m *Manager) consumeStopRequested(id string) bool { // tracked entry for id has already permanently given up (exhausted its // restart attempts — current==nil, gaveUp==true), it is dropped so a fresh // Start can respawn cleanly rather than being rejected as "already -// running" against a dead entry. p.stop() on a gave-up process is a fast, -// safe no-op (its supervision loop has already exited), so this is safe to -// do while holding m.mu. +// running" against a dead entry. The exhausted process is stopped after m.mu +// is released: p.stop() waits for the supervision goroutine, and its final +// status callback may call back into Manager and need the same mutex. func (m *Manager) claimStart(id string) error { m.mu.Lock() - defer m.mu.Unlock() if _, inFlight := m.starting[id]; inFlight { + m.mu.Unlock() return fmt.Errorf("plugins/runtime: plugin %q is already starting", id) } + var exhausted *process if p, exists := m.processes[id]; exists { if !p.isGaveUp() { + m.mu.Unlock() return fmt.Errorf("plugins/runtime: plugin %q is already running", id) } delete(m.processes, id) - p.stop() + exhausted = p } if m.starting == nil { m.starting = make(map[string]struct{}) } m.starting[id] = struct{}{} + m.mu.Unlock() + + if exhausted != nil { + exhausted.stop() + } return nil } diff --git a/apps/backend/internal/plugins/runtime/manager_test.go b/apps/backend/internal/plugins/runtime/manager_test.go index 6cd7eaabc9..c385af974c 100644 --- a/apps/backend/internal/plugins/runtime/manager_test.go +++ b/apps/backend/internal/plugins/runtime/manager_test.go @@ -270,7 +270,7 @@ func TestManager_CrashTriggersAutomaticRestart(t *testing.T) { var mu sync.Mutex var transitions []bool - onStatusChange := func(_ string, healthy bool) { + onStatusChange := func(_ string, healthy bool, _ error) { mu.Lock() transitions = append(transitions, healthy) mu.Unlock() @@ -502,6 +502,60 @@ func TestManager_StartAfterExhaustionRespawns(t *testing.T) { } } +// TestManager_ConcurrentEnableDuringRestartExhaustionDoesNotDeadlock pins the +// lock ordering between the final supervision callback and a manual Enable. +// The callback deliberately re-enters Manager.RestartCount after a barrier; +// Enable must be able to stop the exhausted process without holding m.mu while +// it waits for that callback to return. +func TestManager_ConcurrentEnableDuringRestartExhaustionDoesNotDeadlock(t *testing.T) { + callbackEntered := make(chan struct{}) + releaseCallback := make(chan struct{}) + var m *Manager + m = NewManager(t.TempDir(), func(id string, healthy bool, reason error) { + if healthy || reason == nil || reason.Error() != "plugin restart attempts exhausted" { + return + } + close(callbackEntered) + <-releaseCallback + _ = m.RestartCount(id) + }, testLogger(t)) + t.Cleanup(m.StopAll) + + const id = "deadlock-plugin" + p := newProcess(id, testLogger(t), nil, m.onStatusChange) + p.maxRestartAttempts = 0 + p.current = &fakeSpawnedProcess{} + m.mu.Lock() + m.processes[id] = p + m.mu.Unlock() + + p.wg.Add(1) + go func() { + defer p.wg.Done() + _ = p.handleFailureAndRestart(errors.New("initial failure")) + }() + <-callbackEntered + + startDone := make(chan error, 1) + go func() { + startDone <- m.startProcess(id, func() (spawnedProcess, error) { + return &fakeSpawnedProcess{}, nil + }) + }() + <-p.stopCh + close(releaseCallback) + + select { + case err := <-startDone: + if err != nil { + t.Fatalf("startProcess() returned unexpected error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("concurrent Enable remained blocked after restart-exhaustion callback was released") + } + m.Stop(id) +} + // TestManager_DefaultStartTimeoutIs30s pins the fix for bounded startup: // Manager.Start previously spawned via a hcplugin.Client with no configured // StartTimeout, defaulting to go-plugin's own 1-minute handshake timeout, so diff --git a/apps/backend/internal/plugins/runtime/process.go b/apps/backend/internal/plugins/runtime/process.go index dde02f4cf2..796f756e0c 100644 --- a/apps/backend/internal/plugins/runtime/process.go +++ b/apps/backend/internal/plugins/runtime/process.go @@ -1,6 +1,7 @@ package runtime import ( + "fmt" "sync" "time" @@ -53,7 +54,7 @@ type process struct { log *logger.Logger spawnFn func() (spawnedProcess, error) - onStatusChange func(id string, healthy bool) + onStatusChange func(id string, healthy bool, reason error) pingInterval time.Duration maxConsecutiveFails int @@ -84,7 +85,7 @@ type process struct { // newProcess builds a process with default tuning. Callers may override // pingInterval/maxConsecutiveFails/restartBackoff/maxRestartAttempts/ // sleepFn/onTick before calling start. -func newProcess(id string, log *logger.Logger, spawnFn func() (spawnedProcess, error), onStatusChange func(string, bool)) *process { +func newProcess(id string, log *logger.Logger, spawnFn func() (spawnedProcess, error), onStatusChange func(string, bool, error)) *process { return &process{ id: id, log: log, @@ -227,8 +228,9 @@ func (p *process) tick() bool { } if cur.Exited() { + reason := fmt.Errorf("plugin process exited unexpectedly") p.log.Warn("plugin process exited unexpectedly", zap.String("plugin_id", p.id)) - return p.handleFailureAndRestart() + return p.handleFailureAndRestart(reason) } if err := cur.Ping(); err != nil { @@ -239,7 +241,7 @@ func (p *process) tick() bool { if reachedThreshold { p.log.Warn("plugin health check failed repeatedly", zap.String("plugin_id", p.id), zap.Error(err)) - return p.handleFailureAndRestart() + return p.handleFailureAndRestart(err) } return false } @@ -249,7 +251,7 @@ func (p *process) tick() bool { p.failures = 0 p.mu.Unlock() if hadFailures { - p.notify(true) + p.notify(true, nil) } return false } @@ -258,8 +260,8 @@ func (p *process) tick() bool { // left of the old process, and attempts to respawn it with backoff up to // maxRestartAttempts times. Returns true if every attempt failed (the // process has given up permanently). -func (p *process) handleFailureAndRestart() bool { - p.notify(false) +func (p *process) handleFailureAndRestart(reason error) bool { + p.notify(false, reason) p.mu.Lock() if p.current != nil { @@ -269,6 +271,7 @@ func (p *process) handleFailureAndRestart() bool { p.failures = 0 p.mu.Unlock() + var lastErr error for attempt := 0; attempt < p.maxRestartAttempts; attempt++ { delay := time.Duration(0) if attempt < len(p.restartBackoff) { @@ -280,6 +283,7 @@ func (p *process) handleFailureAndRestart() bool { proc, err := p.spawnFn() if err != nil { + lastErr = err p.log.Warn("plugin restart attempt failed", zap.String("plugin_id", p.id), zap.Int("attempt", attempt+1), zap.Error(err)) continue @@ -289,7 +293,7 @@ func (p *process) handleFailureAndRestart() bool { p.current = proc p.restarts++ p.mu.Unlock() - p.notify(true) + p.notify(true, nil) return false } @@ -298,12 +302,17 @@ func (p *process) handleFailureAndRestart() bool { p.mu.Lock() p.gaveUp = true p.mu.Unlock() + if lastErr != nil { + p.notify(false, fmt.Errorf("plugin restart attempts exhausted: %w", lastErr)) + } else { + p.notify(false, fmt.Errorf("plugin restart attempts exhausted")) + } return true } // notify invokes onStatusChange, if set. -func (p *process) notify(healthy bool) { +func (p *process) notify(healthy bool, reason error) { if p.onStatusChange != nil { - p.onStatusChange(p.id, healthy) + p.onStatusChange(p.id, healthy, reason) } } diff --git a/apps/backend/internal/plugins/runtime/process_test.go b/apps/backend/internal/plugins/runtime/process_test.go index 3e2fc787e5..009eac9ad9 100644 --- a/apps/backend/internal/plugins/runtime/process_test.go +++ b/apps/backend/internal/plugins/runtime/process_test.go @@ -45,17 +45,19 @@ func (f *fakeSpawnedProcess) isKilled() bool { return f.killed } -// statusRecorder records every (id, healthy) pair passed to -// onStatusChange, for assertion. +// statusRecorder records every status transition and its failure cause passed +// to onStatusChange, for assertion. type statusRecorder struct { - mu sync.Mutex - calls []bool + mu sync.Mutex + calls []bool + reasons []error } -func (r *statusRecorder) record(_ string, healthy bool) { +func (r *statusRecorder) record(_ string, healthy bool, reason error) { r.mu.Lock() defer r.mu.Unlock() r.calls = append(r.calls, healthy) + r.reasons = append(r.reasons, reason) } func (r *statusRecorder) snapshot() []bool { @@ -66,6 +68,14 @@ func (r *statusRecorder) snapshot() []bool { return out } +func (r *statusRecorder) snapshotReasons() []error { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]error, len(r.reasons)) + copy(out, r.reasons) + return out +} + // noSleep is a sleepFn that never actually blocks: it returns true // immediately unless stopCh is already closed, in which case it returns // false. Used by every test in this file so supervision-decision logic @@ -171,6 +181,9 @@ func TestProcess_Tick_ThresholdReachedRestartsAndNotifiesUnhealthyThenHealthy(t if want := []bool{false, true}; !equalBoolSlices(rec.snapshot(), want) { t.Fatalf("onStatusChange calls = %v, want %v", rec.snapshot(), want) } + if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "boom" || reasons[1] != nil { + t.Fatalf("onStatusChange reasons = %v, want [boom nil]", reasons) + } } func TestProcess_Tick_ExitedTriggersImmediateRestartRegardlessOfFailureCount(t *testing.T) { @@ -189,6 +202,9 @@ func TestProcess_Tick_ExitedTriggersImmediateRestartRegardlessOfFailureCount(t * if want := []bool{false, true}; !equalBoolSlices(rec.snapshot(), want) { t.Fatalf("onStatusChange calls = %v, want %v (immediate restart on exit)", rec.snapshot(), want) } + if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "plugin process exited unexpectedly" || reasons[1] != nil { + t.Fatalf("onStatusChange reasons = %v, want [process exit nil]", reasons) + } } func TestProcess_HandleFailureAndRestart_GivesUpAfterMaxAttempts(t *testing.T) { @@ -204,14 +220,17 @@ func TestProcess_HandleFailureAndRestart_GivesUpAfterMaxAttempts(t *testing.T) { p.maxRestartAttempts = 3 p.restartBackoff = []time.Duration{0, 0, 0} - if gaveUp := p.handleFailureAndRestart(); !gaveUp { + if gaveUp := p.handleFailureAndRestart(errors.New("initial failure")); !gaveUp { t.Fatal("handleFailureAndRestart() = false, want true after exhausting every attempt") } if spawnCalls != 3 { t.Fatalf("spawnFn called %d times, want 3 (maxRestartAttempts)", spawnCalls) } - if want := []bool{false}; !equalBoolSlices(rec.snapshot(), want) { - t.Fatalf("onStatusChange calls = %v, want %v (only the initial degradation, no false recovery)", rec.snapshot(), want) + if want := []bool{false, false}; !equalBoolSlices(rec.snapshot(), want) { + t.Fatalf("onStatusChange calls = %v, want %v (initial degradation plus final restart failure)", rec.snapshot(), want) + } + if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "initial failure" || reasons[1] == nil || reasons[1].Error() != "plugin restart attempts exhausted: spawn failed" { + t.Fatalf("onStatusChange reasons = %v, want [initial cause, exhausted restart cause]", reasons) } if _, ok := p.remote(); ok { t.Fatal("remote() ok = true after giving up, want false") @@ -231,7 +250,7 @@ func TestProcess_HandleFailureAndRestart_StopDuringBackoffSkipsSpawn(t *testing. close(p.stopCh) // simulate Stop() racing the backoff wait p.sleepFn = sleepOrStop - if gaveUp := p.handleFailureAndRestart(); gaveUp { + if gaveUp := p.handleFailureAndRestart(errors.New("initial failure")); gaveUp { t.Fatal("handleFailureAndRestart() = true, want false: stopping is not the same as giving up") } if spawnCalls != 0 { diff --git a/apps/backend/internal/plugins/service.go b/apps/backend/internal/plugins/service.go index bf07f81df4..e208e0a5a3 100644 --- a/apps/backend/internal/plugins/service.go +++ b/apps/backend/internal/plugins/service.go @@ -665,7 +665,7 @@ func (s *Service) restartForConfigChange(rec *store.Record) error { ctx, cancel := context.WithTimeout(context.Background(), activateStartTimeout) defer cancel() if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil { - if setErr := s.SetStatus(rec.ID, StatusError); setErr != nil { + if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil { // The restart error stays the returned error (it is the primary // signal), but a failed status write means the registry may show // StatusActive with no process running — don't lose that. @@ -1010,15 +1010,17 @@ func (s *Service) Disable(id string) error { const activateStartTimeout = 30 * time.Second // activate spawns rec's process (if not already running) and transitions it -// to StatusActive. If the spawn fails, it best-effort transitions the -// record to StatusError (ignoring an invalid-transition failure, e.g. from -// "disabled") and returns the spawn error. +// to StatusActive. If the spawn fails, it records the failure and transitions +// the record to StatusError before returning the spawn error. func (s *Service) activate(rec *store.Record) error { if s.runtime != nil && !s.runtime.Running(rec.ID) { ctx, cancel := context.WithTimeout(context.Background(), activateStartTimeout) defer cancel() if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil { - _ = s.SetStatus(rec.ID, StatusError) + if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil { + s.log.Warn("plugins: could not persist activation failure", + zap.String("plugin_id", rec.ID), zap.Error(setErr)) + } return fmt.Errorf("plugins: start %q: %w", rec.ID, err) } } @@ -1035,30 +1037,50 @@ func (s *Service) activate(rec *store.Record) error { // both for the runtime spawn/stop and the status transition, and only want // a single Refresh for the whole operation. func (s *Service) SetStatus(id string, status Status) error { + return s.setStatusAndDiagnostic(id, status, nil, false) +} + +// setStatusAndDiagnostic applies a lifecycle transition together with its +// persisted runtime diagnostic. allowSame is used only for repeated failure +// reports (for example, a failed retry while the record is already in error); +// public SetStatus keeps same-state transitions invalid. +func (s *Service) setStatusAndDiagnostic(id string, status Status, failure error, allowSame bool) error { s.mu.Lock() + defer s.mu.Unlock() rec, ok := s.registry.Get(id) if !ok { - s.mu.Unlock() return store.ErrNotFound } - if !canTransition(rec.Status, status) { - s.mu.Unlock() + if rec.Status == status { + if !allowSame { + return &ErrInvalidTransition{ID: id, From: rec.Status, To: status} + } + } else if !canTransition(rec.Status, status) { return &ErrInvalidTransition{ID: id, From: rec.Status, To: status} } - updated, ok := s.registry.SetStatus(id, status) + lastError := rec.LastError + lastErrorAt := rec.LastErrorAt + if status == StatusActive { + lastError = "" + lastErrorAt = nil + } + if failure != nil { + lastError = normalizePluginError(failure) + now := time.Now().UTC() + lastErrorAt = &now + } + + updated, ok := s.registry.SetRuntimeState(id, status, lastError, lastErrorAt) if !ok { - s.mu.Unlock() return store.ErrNotFound } if err := s.store.Save(updated); err != nil { // Roll back the in-memory change so registry and disk stay in sync. - s.registry.SetStatus(id, rec.Status) - s.mu.Unlock() + s.registry.Add(rec) return err } - s.mu.Unlock() return nil } @@ -1068,12 +1090,12 @@ func (s *Service) SetStatus(id string, status Status) error { // health transitions. healthy=false drives active -> error; healthy=true // drives error -> active plus a Deliverer.Flush (the buffered-event // recovery replay). Restart count is persisted best-effort afterward. -func (s *Service) handleStatusChange(id string, healthy bool) { +func (s *Service) handleStatusChange(id string, healthy bool, reason error) { newStatus := StatusError if healthy { newStatus = StatusActive } - if err := s.SetStatus(id, newStatus); err != nil { + if err := s.setStatusAndDiagnostic(id, newStatus, reason, !healthy); err != nil { s.log.Warn("plugins: health transition failed", zap.String("plugin_id", id), zap.Bool("healthy", healthy), zap.Error(err)) } else { @@ -1093,6 +1115,11 @@ func (s *Service) recordRestartCount(id string) { if s.runtime == nil { return } + // Serialize this metadata write with lifecycle/diagnostic persistence so a + // restart callback cannot save a stale record over a concurrent Enable or + // recovery that just cleared LastError. + s.mu.Lock() + defer s.mu.Unlock() updated, ok := s.registry.SetRestartCount(id, s.runtime.RestartCount(id)) if !ok { return @@ -1122,7 +1149,15 @@ func (s *Service) StartActivePlugins(ctx context.Context) { if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil { s.log.Warn("plugins: failed to spawn active plugin at boot", zap.String("plugin_id", rec.ID), zap.Error(err)) - _ = s.SetStatus(rec.ID, StatusError) + if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil { + s.log.Warn("plugins: could not persist boot activation failure", + zap.String("plugin_id", rec.ID), zap.Error(setErr)) + } else { + // The deliverer was refreshed before boot activation began, so + // reconcile it after an active plugin fails to spawn. Otherwise + // its worker would continue treating the plugin as active. + s.notifyDeliverer() + } } } } diff --git a/apps/backend/internal/plugins/service_sync.go b/apps/backend/internal/plugins/service_sync.go index 42ac064f29..4505d9c6d5 100644 --- a/apps/backend/internal/plugins/service_sync.go +++ b/apps/backend/internal/plugins/service_sync.go @@ -2,7 +2,6 @@ package plugins import ( "context" - "errors" "fmt" "os" "path/filepath" @@ -274,6 +273,12 @@ func (s *Service) scanMissingInstalls() []string { } if _, err := os.Stat(rec.InstallPath); err == nil { continue + } else if !os.IsNotExist(err) { + s.log.Warn("plugins: sync could not inspect install path", + zap.String("plugin_id", rec.ID), + zap.String("path", rec.InstallPath), + zap.Error(err)) + continue } if s.runtime != nil && s.runtime.Running(rec.ID) { s.runtime.Stop(rec.ID) @@ -284,35 +289,22 @@ func (s *Service) scanMissingInstalls() []string { return missing } -// markMissing transitions id to StatusError via the normal FSM transition, -// falling back to a direct (transition-check-bypassing) status write if the -// transition is not a legal single-hop edge — e.g. the record is already -// StatusError, which SetStatus rejects as a same-status "transition". +// markMissing transitions id to StatusError and records the missing path as +// the current runtime diagnostic. Repeated scans replace the timestamp and +// diagnostic instead of bypassing the normal persistence path. func (s *Service) markMissing(id string) { - err := s.SetStatus(id, StatusError) + rec, getErr := s.Get(id) + if getErr != nil { + s.log.Warn("plugins: sync failed to read missing install record", + zap.String("plugin_id", id), zap.Error(getErr)) + return + } + err := s.setStatusAndDiagnostic(id, StatusError, + fmt.Errorf("plugin install path %q is missing", rec.InstallPath), true) if err == nil { s.notifyDeliverer() return } - var invalidErr *ErrInvalidTransition - if !errors.As(err, &invalidErr) { - s.log.Warn("plugins: sync failed to mark missing install as error", - zap.String("plugin_id", id), zap.Error(err)) - return - } - if err := s.setStatusUnchecked(id, StatusError); err != nil { - s.log.Warn("plugins: sync failed to force missing install to error", - zap.String("plugin_id", id), zap.Error(err)) - } -} - -// setStatusUnchecked directly writes status onto id's record in both the -// registry and the store, bypassing SetStatus's FSM transition check. Only -// used by markMissing's fallback path above. -func (s *Service) setStatusUnchecked(id string, status Status) error { - updated, ok := s.registry.SetStatus(id, status) - if !ok { - return store.ErrNotFound - } - return s.store.Save(updated) + s.log.Warn("plugins: sync failed to mark missing install as error", + zap.String("plugin_id", id), zap.Error(err)) } diff --git a/apps/backend/internal/plugins/service_sync_test.go b/apps/backend/internal/plugins/service_sync_test.go index 8086c379f7..ca71793834 100644 --- a/apps/backend/internal/plugins/service_sync_test.go +++ b/apps/backend/internal/plugins/service_sync_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" goruntime "runtime" + "strings" "testing" "time" ) @@ -217,6 +218,31 @@ func TestServiceSync_MissingInstallPathSetsErrorAndStopsRuntime(t *testing.T) { if got.Status != StatusError { t.Fatalf("Status after missing-install detection = %q, want %q", got.Status, StatusError) } + if got.LastError == "" || got.LastErrorAt == nil || !strings.Contains(got.LastError, "install path") { + t.Fatalf("diagnostic after missing-install detection = (%q, %v), want install path/non-nil", got.LastError, got.LastErrorAt) + } +} + +func TestServiceSync_StatErrorDoesNotMarkInstallMissing(t *testing.T) { + svc, _, _ := newTestService(t) + rec := installTestPlugin(t, svc, "kandev-plugin-stat-error") + rec.InstallPath = "\x00" + svc.registry.Add(rec) + + result, err := svc.Sync(context.Background()) + if err != nil { + t.Fatalf("Sync() unexpected error: %v", err) + } + if len(result.Missing) != 0 { + t.Fatalf("Sync().Missing = %v, want none for a non-NotExist stat error", result.Missing) + } + got, err := svc.Get("kandev-plugin-stat-error") + if err != nil { + t.Fatalf("Get(): %v", err) + } + if got.Status != StatusActive { + t.Fatalf("Status after stat error = %q, want %q", got.Status, StatusActive) + } } // TestServiceSync_MissingInstallAlreadyErrorIsIdempotent proves the "direct diff --git a/apps/backend/internal/plugins/service_test.go b/apps/backend/internal/plugins/service_test.go index 74860bf0d5..2a6a485395 100644 --- a/apps/backend/internal/plugins/service_test.go +++ b/apps/backend/internal/plugins/service_test.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" goruntime "runtime" + "strings" "sync" "testing" "time" @@ -148,6 +149,12 @@ func (r *fakeRuntime) setStartErr(id string, err error) { r.startErr[id] = err } +func (r *fakeRuntime) clearStartErr(id string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.startErr, id) +} + func (r *fakeRuntime) startCallCount(id string) int { r.mu.Lock() defer r.mu.Unlock() @@ -489,6 +496,81 @@ func TestServiceDisabledCanReEnableAndRespawns(t *testing.T) { } } +func TestServiceEnableFailurePersistsDiagnosticAndSuccessfulRetryClearsIt(t *testing.T) { + svc, fsStore, rt := newTestService(t) + installTestPlugin(t, svc, "kandev-plugin-slack") + if err := svc.Disable("kandev-plugin-slack"); err != nil { + t.Fatalf("Disable(): %v", err) + } + + rt.setStartErr("kandev-plugin-slack", errors.New("handshake failed: binary exited")) + if err := svc.Enable("kandev-plugin-slack"); err == nil { + t.Fatal("Enable() unexpectedly succeeded with a configured start failure") + } + + failed, err := svc.Get("kandev-plugin-slack") + if err != nil { + t.Fatalf("Get() after failed Enable(): %v", err) + } + if failed.Status != StatusError { + t.Fatalf("Status after failed Enable() = %q, want %q", failed.Status, StatusError) + } + if failed.LastError == "" || !strings.Contains(failed.LastError, "handshake failed") { + t.Fatalf("LastError = %q, want the start failure", failed.LastError) + } + if failed.LastErrorAt == nil { + t.Fatal("LastErrorAt is nil after failed Enable()") + } + onDisk, err := fsStore.Get("kandev-plugin-slack") + if err != nil { + t.Fatalf("store.Get() after failed Enable(): %v", err) + } + if onDisk.LastError != failed.LastError || onDisk.LastErrorAt == nil { + t.Fatalf("persisted diagnostic = (%q, %v), want (%q, non-nil)", onDisk.LastError, onDisk.LastErrorAt, failed.LastError) + } + + rt.clearStartErr("kandev-plugin-slack") + if err := svc.Enable("kandev-plugin-slack"); err != nil { + t.Fatalf("Enable() retry unexpected error: %v", err) + } + recovered, err := svc.Get("kandev-plugin-slack") + if err != nil { + t.Fatalf("Get() after successful retry: %v", err) + } + if recovered.Status != StatusActive { + t.Fatalf("Status after successful retry = %q, want %q", recovered.Status, StatusActive) + } + if recovered.LastError != "" || recovered.LastErrorAt != nil { + t.Fatalf("diagnostic after successful retry = (%q, %v), want empty/nil", recovered.LastError, recovered.LastErrorAt) + } +} + +func TestServiceFailedRetryReplacesDiagnostic(t *testing.T) { + svc, _, rt := newTestService(t) + installTestPlugin(t, svc, "kandev-plugin-slack") + if err := svc.Disable("kandev-plugin-slack"); err != nil { + t.Fatalf("Disable(): %v", err) + } + + rt.setStartErr("kandev-plugin-slack", errors.New("first handshake failure")) + if err := svc.Enable("kandev-plugin-slack"); err == nil { + t.Fatal("first Enable() unexpectedly succeeded") + } + first, _ := svc.Get("kandev-plugin-slack") + + rt.setStartErr("kandev-plugin-slack", errors.New("second executable failure")) + if err := svc.Enable("kandev-plugin-slack"); err == nil { + t.Fatal("second Enable() unexpectedly succeeded") + } + second, _ := svc.Get("kandev-plugin-slack") + if second.Status != StatusError { + t.Fatalf("Status after failed retry = %q, want %q", second.Status, StatusError) + } + if second.LastError == first.LastError || !strings.Contains(second.LastError, "second executable failure") { + t.Fatalf("LastError after failed retry = %q, want replacement diagnostic", second.LastError) + } +} + // TestServiceEnable_ConcurrentCallsForSameID_OnlyOneActivationNoError proves // Enable is guarded by a per-plugin lock: two near-simultaneous Enable calls // for the same disabled plugin id must not both pass the StatusActive @@ -588,12 +670,15 @@ func TestServiceHandleStatusChangeUnhealthyTransitionsToErrorAndRefreshesDeliver deliverer := &fakeDeliverer{} svc.SetDeliverer(deliverer) - svc.handleStatusChange("kandev-plugin-slack", false) + svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout")) got, _ := svc.Get("kandev-plugin-slack") if got.Status != StatusError { t.Fatalf("Status after unhealthy transition = %q, want %q", got.Status, StatusError) } + if got.LastError == "" || !strings.Contains(got.LastError, "ping timeout") || got.LastErrorAt == nil { + t.Fatalf("diagnostic after unhealthy transition = (%q, %v), want ping timeout/non-nil", got.LastError, got.LastErrorAt) + } if deliverer.refreshCount != 1 { t.Fatalf("Refresh() call count = %d, want 1", deliverer.refreshCount) } @@ -605,16 +690,19 @@ func TestServiceHandleStatusChangeUnhealthyTransitionsToErrorAndRefreshesDeliver func TestServiceHandleStatusChangeHealthyRecoversAndFlushesDeliverer(t *testing.T) { svc, _, _ := newTestService(t) installTestPlugin(t, svc, "kandev-plugin-slack") - svc.handleStatusChange("kandev-plugin-slack", false) // degrade first + svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout")) // degrade first deliverer := &fakeDeliverer{} svc.SetDeliverer(deliverer) - svc.handleStatusChange("kandev-plugin-slack", true) + svc.handleStatusChange("kandev-plugin-slack", true, nil) got, _ := svc.Get("kandev-plugin-slack") if got.Status != StatusActive { t.Fatalf("Status after recovery = %q, want %q", got.Status, StatusActive) } + if got.LastError != "" || got.LastErrorAt != nil { + t.Fatalf("diagnostic after recovery = (%q, %v), want empty/nil", got.LastError, got.LastErrorAt) + } if len(deliverer.flushedIDs) != 1 || deliverer.flushedIDs[0] != "kandev-plugin-slack" { t.Fatalf("Flush() calls = %v, want [kandev-plugin-slack]", deliverer.flushedIDs) } @@ -627,7 +715,7 @@ func TestServiceHandleStatusChangePersistsRestartCountBestEffort(t *testing.T) { rt.restartCounts["kandev-plugin-slack"] = 3 rt.mu.Unlock() - svc.handleStatusChange("kandev-plugin-slack", false) + svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout")) got, _ := svc.Get("kandev-plugin-slack") if got.RestartCount != 3 { @@ -780,3 +868,38 @@ func TestServiceStartActivePluginsSpawnsOnlyActiveManagedNotAlreadyRunning(t *te t.Fatal("StartActivePlugins() did not spawn the active plugin") } } + +func TestServiceStartActivePluginsFailurePersistsDiagnosticAndRefreshesDeliverer(t *testing.T) { + svc, dir, fsStore, _ := newTestServiceWithDir(t) + installTestPlugin(t, svc, "kandev-plugin-slack") + + reg2 := NewRegistry() + if err := reg2.Load(fsStore); err != nil { + t.Fatalf("Load(): %v", err) + } + svc2 := NewService(fsStore, reg2, nil, testLogger(t)) + svc2.SetPluginsDir(dir) + rt2 := newFakeRuntime() + rt2.setStartErr("kandev-plugin-slack", errors.New("boot handshake failed")) + svc2.SetRuntime(rt2) + deliverer := &fakeDeliverer{} + svc2.SetDeliverer(deliverer) + + svc2.StartActivePlugins(context.Background()) + + got, err := svc2.Get("kandev-plugin-slack") + if err != nil { + t.Fatalf("Get() after boot failure: %v", err) + } + if got.Status != StatusError { + t.Fatalf("Status after boot failure = %q, want %q", got.Status, StatusError) + } + if !strings.Contains(got.LastError, "boot handshake failed") || got.LastErrorAt == nil { + t.Fatalf("diagnostic after boot failure = (%q, %v), want boot failure and timestamp", got.LastError, got.LastErrorAt) + } + // bootScan refreshes once at the end of its normal reconciliation, then + // the failed active spawn refreshes again so delivery sees StatusError. + if deliverer.refreshCount != 2 { + t.Fatalf("Refresh() calls after boot failure = %d, want 2", deliverer.refreshCount) + } +} diff --git a/apps/backend/internal/plugins/store/fs_store_test.go b/apps/backend/internal/plugins/store/fs_store_test.go index abdd18d710..75c9f13e6f 100644 --- a/apps/backend/internal/plugins/store/fs_store_test.go +++ b/apps/backend/internal/plugins/store/fs_store_test.go @@ -216,6 +216,51 @@ func TestFSStore_Save_PersistsUpdatedRecord(t *testing.T) { } } +func TestFSStore_Save_PersistsFailureDiagnostics(t *testing.T) { + dir := t.TempDir() + s := NewFSStore(dir) + + failedAt := time.Date(2026, time.August, 2, 12, 34, 56, 0, time.UTC) + rec := testRecord("kandev-plugin-slack") + rec.Status = StatusError + rec.LastError = "plugins/runtime: connect to plugin: handshake failed" + rec.LastErrorAt = &failedAt + if err := s.Save(rec); err != nil { + t.Fatalf("Save() unexpected error: %v", err) + } + + got, err := s.Get(rec.ID) + if err != nil { + t.Fatalf("Get() unexpected error: %v", err) + } + if got.LastError != rec.LastError { + t.Fatalf("Get().LastError = %q, want %q", got.LastError, rec.LastError) + } + if got.LastErrorAt == nil || !got.LastErrorAt.Equal(failedAt) { + t.Fatalf("Get().LastErrorAt = %v, want %v", got.LastErrorAt, failedAt) + } +} + +func TestFSStore_Get_LegacyRecordDefaultsFailureDiagnostics(t *testing.T) { + dir := t.TempDir() + s := NewFSStore(dir) + rec := testRecord("kandev-plugin-slack") + if err := s.Save(rec); err != nil { + t.Fatalf("Save() unexpected error: %v", err) + } + + got, err := s.Get(rec.ID) + if err != nil { + t.Fatalf("Get() unexpected error: %v", err) + } + if got.LastError != "" { + t.Fatalf("legacy record LastError = %q, want empty", got.LastError) + } + if got.LastErrorAt != nil { + t.Fatalf("legacy record LastErrorAt = %v, want nil", got.LastErrorAt) + } +} + func TestFSStore_SetConfigThenGetConfig_RoundTrips(t *testing.T) { dir := t.TempDir() s := NewFSStore(dir) diff --git a/apps/backend/internal/plugins/store/store.go b/apps/backend/internal/plugins/store/store.go index 6782414560..e1bd5db25d 100644 --- a/apps/backend/internal/plugins/store/store.go +++ b/apps/backend/internal/plugins/store/store.go @@ -48,9 +48,11 @@ type Record struct { // and false forces it off — either override wins over the global default. // Persisted so the choice survives restart and, unlike Status, is carried // over verbatim when a plugin is upgraded in place (see Service.Install). - AutoUpdate *bool `yaml:"auto_update,omitempty" json:"auto_update,omitempty"` - InstalledAt time.Time `yaml:"installed_at" json:"installed_at"` - RestartCount int `yaml:"restart_count" json:"restart_count"` + AutoUpdate *bool `yaml:"auto_update,omitempty" json:"auto_update,omitempty"` + InstalledAt time.Time `yaml:"installed_at" json:"installed_at"` + RestartCount int `yaml:"restart_count" json:"restart_count"` + LastError string `yaml:"last_error,omitempty" json:"last_error,omitempty"` + LastErrorAt *time.Time `yaml:"last_error_at,omitempty" json:"last_error_at,omitempty"` } // Store is the persistence interface for plugin installations and their diff --git a/apps/backend/internal/plugins/types.go b/apps/backend/internal/plugins/types.go index 58568afbf3..6ac2c34927 100644 --- a/apps/backend/internal/plugins/types.go +++ b/apps/backend/internal/plugins/types.go @@ -18,8 +18,8 @@ import ( // docs/specs/plugins/spec.md ("State machine"): // // registered -> active -> disabled -> uninstalled -// | | -// +-> error -+ +// registered|active|disabled --failure--> error +// recovery --> active; Disable --> disabled // // Status is a type alias (not a distinct type) for store's Record.Status // field, and the constants below are direct aliases of the string values @@ -60,7 +60,7 @@ var allowedTransitions = map[Status][]Status{ StatusRegistered: {StatusActive, StatusError}, StatusActive: {StatusDisabled, StatusError}, StatusError: {StatusActive, StatusDisabled}, - StatusDisabled: {StatusActive}, + StatusDisabled: {StatusActive, StatusError}, } // canTransition reports whether to is a legal single-hop transition from diff --git a/apps/web/components/settings/plugins/plugin-detail.tsx b/apps/web/components/settings/plugins/plugin-detail.tsx index 324324bf25..de520fb756 100644 --- a/apps/web/components/settings/plugins/plugin-detail.tsx +++ b/apps/web/components/settings/plugins/plugin-detail.tsx @@ -15,6 +15,7 @@ import { PluginConfigForm } from "./plugin-config-form"; import { PluginManifestCard } from "./plugin-manifest-card"; import { PluginRepoLink } from "./plugin-repo-link"; import { PluginStatusBadge } from "./plugin-status-badge"; +import { PluginErrorDiagnostic } from "./plugin-error-diagnostic"; import { UninstallPluginDialog } from "./uninstall-plugin-dialog"; import { usePluginActions } from "./use-plugin-actions"; import { usePluginConfigForm } from "./use-plugin-config-form"; @@ -114,6 +115,7 @@ function PluginDetailHeader({ plugin }: PluginDetailHeaderProps) { {plugin.description && (

{plugin.description}

)} + @@ -178,7 +180,8 @@ type PluginDangerZoneProps = { function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) { const busy = actions.busyId === plugin.id; - const canEnable = plugin.status === "disabled" || plugin.status === "registered"; + const canEnable = + plugin.status === "disabled" || plugin.status === "registered" || plugin.status === "error"; const canDisable = plugin.status === "active" || plugin.status === "error"; return ( @@ -187,7 +190,7 @@ function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) {