From 5c93f2651784ec5786cc48c8614f010ed5745673 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=BB=95=E5=8B=87=E5=BF=97?= <12671205+Qsnh@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:40:00 -0700 Subject: [PATCH 1/2] fix(config): add missing closing brace to WeComConfig WeComConfig was missing its closing brace, so TelegramConfig was parsed as part of its field list and internal/infra/config failed to compile: config.go:185:1: syntax error: unexpected keyword type, expected field name or embedded type This broke `go build ./...` on main. Co-Authored-By: Claude Opus 5 --- internal/infra/config/config.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 0949b086..e017046c 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -181,6 +181,7 @@ type WeComConfig struct { Secret string `yaml:"secret"` WebSocketURL string `yaml:"websocket_url"` BotName string `yaml:"bot_name"` +} type TelegramConfig struct { Enabled bool `yaml:"enabled"` From dcedb8d346798db179e26e58875703d3e8386a5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=BB=95=E5=8B=87=E5=BF=97?= <12671205+Qsnh@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:40:09 -0700 Subject: [PATCH 2/2] fix(app): drain background runners before closing the database App.Run cancelled the runner context and then closed the database without waiting for any of the eight fire-and-forget goroutines. The feeder, scheduler, dispatcher, reconciler and token-stats syncer all touch the database on their way out of a cancelled loop, so shutdown raced against db.Close. Rebuild Run around errgroup, already the concurrency idiom used elsewhere in this repo. `defer a.db.Close()` is declared before g.Wait(), so it unwinds only after every runner has returned. A server error now cancels the group and propagates out of Run, so `openbee server` exits non-zero instead of logging and returning success. Also folded in, all in the same shutdown path: - Build the platform sender map in one shot (buildSenders) instead of handing an empty map to the failure notifier and filling it afterwards. - Move startup recovery out of BuildApp into App.run, so constructing an App no longer mutates the database. - Extract the hidden os.Setenv("OPENBEE_URL") out of buildAllEngines into publishRPCBaseURL. - Guard routes.Server.httpServer with a mutex; Run wrote it while Shutdown read it from another goroutine. - Name each runner so exits are traceable in logs. Tests drive App.run directly through an httpServer stub. Verified to fail against the previous behaviour before being committed. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 + cmd/openbee/internal/cli/daemoncmd/server.go | 3 +- internal/app/app.go | 201 ++++++++++----- internal/app/lifecycle_test.go | 243 +++++++++++++++++++ internal/routes/server.go | 22 +- 5 files changed, 404 insertions(+), 71 deletions(-) create mode 100644 internal/app/lifecycle_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c3eed1a2..427c99b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,13 @@ ## [Unreleased] +### Changed +- `openbee server` now exits with a non-zero status when the HTTP server fails to start, for example when the port is already in use. Previously the error was logged and the process still exited successfully. + ### Fixed +- Wait for every background loop (message gateways, platform receivers, feeder, scheduler, dispatcher, reconciler and token-stats syncer) to finish before closing the database on shutdown. Previously the database was closed as soon as the HTTP server stopped, while those loops were still writing to it. +- Build the platform sender map completely before passing it to the task failure notifier, removing a startup-order dependency on mutating a shared map after handing out a reference to it. +- Guard the HTTP server handle so that `Run` and `Shutdown` no longer race when shutdown begins while the listener is still starting. - Fix `openbee ctl` failing with `unauthorized` for the remainder of a long-running worker execution. The worker token is minted once at process launch, so a TTL shorter than the execution left every subsequent call rejected; the default `bee.rpc.token_ttl` is now 48h instead of 2h. ## [0.0.42] - 2026-07-01 diff --git a/cmd/openbee/internal/cli/daemoncmd/server.go b/cmd/openbee/internal/cli/daemoncmd/server.go index ed41379c..e53d3b02 100644 --- a/cmd/openbee/internal/cli/daemoncmd/server.go +++ b/cmd/openbee/internal/cli/daemoncmd/server.go @@ -67,8 +67,7 @@ func NewServerCommand() *cobra.Command { return fmt.Errorf("build app: %w", err) } - a.Run() - return nil + return a.Run() }, } cmd.Flags().StringVarP(&cfgPath, "config", "c", "config.yaml", i18n.M.Flag.ConfigPath) diff --git a/internal/app/app.go b/internal/app/app.go index 28478936..a16cf1cf 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -16,6 +16,7 @@ import ( "github.com/theopenbee/openbee/internal/infra/auth" "github.com/theopenbee/openbee/internal/routes" "go.uber.org/zap" + "golang.org/x/sync/errgroup" ai "github.com/theopenbee/openbee/internal/ai" _ "github.com/theopenbee/openbee/internal/ai/claude" @@ -48,51 +49,87 @@ import ( webui "github.com/theopenbee/openbee/web" ) +// shutdownTimeout bounds how long the HTTP server gets to finish in-flight +// requests once shutdown begins. +const shutdownTimeout = 15 * time.Second + +// runner is a named long-lived background loop. run must return when ctx is +// cancelled; Run waits for every one of them before closing the database. +type runner struct { + name string + run func(ctx context.Context) +} + +// httpServer is the slice of *routes.Server that App depends on. Keeping it an +// interface lets the lifecycle be tested without standing up the full router. +type httpServer interface { + Run(addr string) error + Shutdown(ctx context.Context) error +} + // App holds all wired-up components and runs the server. type App struct { db *sql.DB - server *routes.Server - runners []func(ctx context.Context) + server httpServer + runners []runner addr string + + // recoverInflight re-hydrates work left behind by a previous process. It + // runs at the start of Run rather than during BuildApp, so constructing an + // App has no side effects. + recoverInflight func(ctx context.Context) } -// Run starts all goroutines, waits for a signal, then shuts down gracefully. -func (a *App) Run() { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() +// Run starts all background runners and the HTTP server, then blocks until +// SIGINT/SIGTERM or an unrecoverable server error. +func (a *App) Run() error { + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + return a.run(ctx) +} + +// run is Run without signal handling, so tests can drive shutdown directly. +func (a *App) run(ctx context.Context) error { + // Deferred first, so it unwinds last — after g.Wait() below has confirmed + // every runner returned. Runners touch the database on their way out of a + // cancelled loop, so closing it any earlier is a race. + defer a.db.Close() + + a.recoverInflight(ctx) + + g, gctx := errgroup.WithContext(ctx) for _, r := range a.runners { - r := r - go r(ctx) + g.Go(func() error { + r.run(gctx) + logger.Debug("runner exited", zap.String("runner", r.name)) + return nil + }) } - serverErr := make(chan error, 1) - go func() { + g.Go(func() error { logger.Info("OpenBee Core starting", zap.String("addr", a.addr)) + // A non-nil return here cancels gctx, which drains every runner. if err := a.server.Run(a.addr); err != nil && !errors.Is(err, http.ErrServerClosed) { - serverErr <- err + return fmt.Errorf("http server: %w", err) } - }() - - quit := make(chan os.Signal, 1) - signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + return nil + }) - select { - case <-quit: + g.Go(func() error { + <-gctx.Done() logger.Info("Shutting down...") - case err := <-serverErr: - logger.Error("server error", zap.Error(err)) - } - - cancel() - - shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 15*time.Second) - defer shutdownCancel() - if err := a.server.Shutdown(shutdownCtx); err != nil { - logger.Error("server shutdown error", zap.Error(err)) - } + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) + defer cancel() + if err := a.server.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("http server shutdown: %w", err) + } + return nil + }) - a.db.Close() + err := g.Wait() + logger.Info("all runners drained") + return err } // BuildApp wires all components together. Returns a ready-to-run App. @@ -106,6 +143,10 @@ func BuildApp(cfg config.Config) (*App, error) { return nil, err } + if err := publishRPCBaseURL(cfg.Bee); err != nil { + return nil, err + } + engines, err := buildAllEngines(cfg.Bee) if err != nil { return nil, fmt.Errorf("init engines: %w", err) @@ -137,18 +178,11 @@ func BuildApp(cfg config.Config) (*App, error) { dispatchCh := make(chan task.DispatchTask, 128) - sendersByPlatform := make(map[string]platform.PlatformSenderAdapter) - - // sendersByPlatform is populated below; notifier holds a reference to the same map. - failureNotifier := task.NewPlatformFailureNotifier(s.msgStore, sendersByPlatform) - feeder, sched := buildBee(cfg.Bee, s, dispatchCh, failureNotifier, engines, engineCfg, envSvc) - // Local platform — always enabled, separate gateway with short debounce localHub := local.NewSSEHub() localReceiver := local.NewLocalReceiver(64) rawLocalSender := local.NewLocalSender(localHub) localSender := store.NewLoggingPlatformSenderAdapter(rawLocalSender, s.outboundMsgStore, local.PlatformID) - sendersByPlatform[local.PlatformID] = localSender platforms, err := buildPlatforms( cfg.Bee.Platforms.Feishu, cfg.Bee.Platforms.DingTalk, cfg.Bee.Platforms.WeCom, @@ -158,9 +192,13 @@ func BuildApp(cfg config.Config) (*App, error) { if err != nil { return nil, err } - for _, p := range platforms { - sendersByPlatform[p.ID()] = store.NewLoggingPlatformSenderAdapter(p.Sender(), s.outboundMsgStore, p.ID()) - } + + // Built in one shot, before any consumer sees it. Nothing may capture a + // reference to this map while it is still being filled. + sendersByPlatform := buildSenders(platforms, localSender, s.outboundMsgStore) + + failureNotifier := task.NewPlatformFailureNotifier(s.msgStore, sendersByPlatform) + feeder, sched := buildBee(cfg.Bee, s, dispatchCh, failureNotifier, engines, engineCfg, envSvc) disp := buildDispatcher(s, mgr, dispatchCh, failureNotifier, engineCfg) beeBusy := command.NewBeeBusyChecker(s.msgStore, s.execStore) @@ -200,37 +238,43 @@ func BuildApp(cfg config.Config) (*App, error) { beeRPCSrv := rpc.NewBeeServer(s.workerStore, mgr, s.taskStore, s.msgStore, s.outboundMsgStore, sendersByPlatform, clearSvc, s.execStore, s.constraintStore, s.sessionStore, s.departmentStore) - // Synchronous startup recovery — must run before goroutines start - feeder.RecoverFeeding(context.Background()) - sched.RecoverRunning(context.Background()) - if n, err := s.execStore.ResetRunningExecutions(context.Background()); err != nil { - logger.Error("recover running executions", zap.Error(err)) - } else if n > 0 { - logger.Info("reset orphaned executions", zap.Int64("count", n)) + // Startup recovery is deferred to Run so that building an App stays free of + // side effects. It still completes before any runner starts. + recoverInflight := func(ctx context.Context) { + feeder.RecoverFeeding(ctx) + sched.RecoverRunning(ctx) + if n, err := s.execStore.ResetRunningExecutions(ctx); err != nil { + logger.Error("recover running executions", zap.Error(err)) + } else if n > 0 { + logger.Info("reset orphaned executions", zap.Int64("count", n)) + } } tokenSyncer := tokenstat.NewSyncer(db, s.tokenStatsStore, engines, ai.AllEngines()) reconciler := task.NewReconciler(s.taskStore, s.execStore, 0) - runners := []func(ctx context.Context){ - func(ctx context.Context) { ingest.Run(ctx) }, - func(ctx context.Context) { localIngest.Run(ctx) }, - func(ctx context.Context) { + runners := []runner{ + {name: "ingest", run: ingest.Run}, + {name: "ingest:local", run: localIngest.Run}, + {name: "receiver:local", run: func(ctx context.Context) { if err := localReceiver.Start(ctx, localIngest.Dispatch); err != nil { logger.Error("local receiver error", zap.Error(err)) } - }, - func(ctx context.Context) { feeder.Run(ctx) }, - func(ctx context.Context) { sched.Run(ctx) }, - func(ctx context.Context) { disp.Run(ctx) }, - func(ctx context.Context) { reconciler.Run(ctx) }, - func(ctx context.Context) { tokenSyncer.Run(ctx) }, + }}, + {name: "feeder", run: feeder.Run}, + {name: "scheduler", run: sched.Run}, + {name: "dispatcher", run: disp.Run}, + {name: "reconciler", run: reconciler.Run}, + {name: "tokenstat", run: tokenSyncer.Run}, } for _, p := range platforms { - recv := p.Receiver() - runners = append(runners, func(ctx context.Context) { - if err := recv.Start(ctx, ingest.Dispatch); err != nil { - logger.Error("platform receiver error", zap.Error(err)) - } + recv, id := p.Receiver(), p.ID() + runners = append(runners, runner{ + name: "receiver:" + id, + run: func(ctx context.Context) { + if err := recv.Start(ctx, ingest.Dispatch); err != nil { + logger.Error("platform receiver error", zap.String("platform", id), zap.Error(err)) + } + }, }) } @@ -246,7 +290,28 @@ func BuildApp(cfg config.Config) (*App, error) { } addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port) - return &App{db: db, server: srv, runners: runners, addr: addr}, nil + return &App{ + db: db, + server: srv, + runners: runners, + addr: addr, + recoverInflight: recoverInflight, + }, nil +} + +// buildSenders returns the complete platform-id -> sender map. +func buildSenders( + platforms []platform.Platform, + localSender platform.PlatformSenderAdapter, + out *store.OutboundMessageStore, +) map[string]platform.PlatformSenderAdapter { + senders := map[string]platform.PlatformSenderAdapter{ + local.PlatformID: localSender, + } + for _, p := range platforms { + senders[p.ID()] = store.NewLoggingPlatformSenderAdapter(p.Sender(), out, p.ID()) + } + return senders } // appStores groups all store instances for passing to sub-builders. @@ -291,10 +356,18 @@ func buildStores(cfg config.DatabaseConfig) (*sql.DB, appStores, error) { }, nil } +// publishRPCBaseURL exports the RPC base URL to the process environment. +// Engine CLIs (claude/codex/pi) read OPENBEE_URL from their inherited env, so +// this must run before any engine adapter is constructed. +func publishRPCBaseURL(cfg config.BeeConfig) error { + if err := os.Setenv("OPENBEE_URL", cfg.RPCBaseURL); err != nil { + return fmt.Errorf("publish OPENBEE_URL: %w", err) + } + return nil +} + // buildAllEngines initializes engine adapters shared safely across concurrent workers. func buildAllEngines(cfg config.BeeConfig) (map[string]ai.EngineAdapter, error) { - os.Setenv("OPENBEE_URL", cfg.RPCBaseURL) //nolint:errcheck - result := make(map[string]ai.EngineAdapter) for _, name := range ai.AllEngines() { if !cfg.Engines.IsEnabled(name) { diff --git a/internal/app/lifecycle_test.go b/internal/app/lifecycle_test.go new file mode 100644 index 00000000..57cbd1c4 --- /dev/null +++ b/internal/app/lifecycle_test.go @@ -0,0 +1,243 @@ +package app + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/theopenbee/openbee/internal/infra/store" +) + +// stubServer stands in for *routes.Server so the lifecycle can be exercised +// without building the full router. Run blocks until Shutdown is called, +// mirroring http.Server semantics. +type stubServer struct { + started chan struct{} + stopped chan struct{} + stopOnce sync.Once + runErr error +} + +func newStubServer(runErr error) *stubServer { + return &stubServer{ + started: make(chan struct{}), + stopped: make(chan struct{}), + runErr: runErr, + } +} + +func (s *stubServer) Run(string) error { + close(s.started) + if s.runErr != nil { + return s.runErr + } + <-s.stopped + return nil +} + +func (s *stubServer) Shutdown(context.Context) error { + s.stopOnce.Do(func() { close(s.stopped) }) + return nil +} + +func (s *stubServer) waitStarted(t *testing.T) { + t.Helper() + select { + case <-s.started: + case <-time.After(5 * time.Second): + t.Fatal("server never started") + } +} + +func newTestDB(t *testing.T) *sql.DB { + t.Helper() + db, err := store.InitDB(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("init test database: %v", err) + } + return db +} + +// runInBackground starts a.run and returns a channel carrying its result. +func runInBackground(a *App, ctx context.Context) <-chan error { + done := make(chan error, 1) + go func() { done <- a.run(ctx) }() + return done +} + +func awaitRun(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(10 * time.Second): + t.Fatal("run did not return after shutdown") + return nil + } +} + +// The regression this whole change exists for: the database must stay open +// until every runner has returned. The probe deliberately lingers after +// cancellation and then touches the database — a premature Close surfaces +// here as "sql: database is closed". +func TestRunClosesDatabaseAfterRunnersDrain(t *testing.T) { + db := newTestDB(t) + var probeErr atomic.Value + // probeDone is what makes this a real regression test: without it, run() + // returning early would leave probeErr unset and the assertions would pass + // vacuously. + var probeDone atomic.Bool + + probe := runner{name: "db-probe", run: func(ctx context.Context) { + <-ctx.Done() + time.Sleep(100 * time.Millisecond) + var n int + if err := db.QueryRow("SELECT 1").Scan(&n); err != nil { + probeErr.Store(err) + } + probeDone.Store(true) + }} + + srv := newStubServer(nil) + a := &App{ + db: db, + server: srv, + runners: []runner{probe}, + addr: "127.0.0.1:0", + recoverInflight: func(context.Context) {}, + } + + ctx, cancel := context.WithCancel(context.Background()) + done := runInBackground(a, ctx) + srv.waitStarted(t) + cancel() + + if err := awaitRun(t, done); err != nil { + t.Fatalf("run returned an error: %v", err) + } + if !probeDone.Load() { + t.Fatal("run returned before the runner finished; the database was closed mid-flight") + } + if err, ok := probeErr.Load().(error); ok && err != nil { + t.Fatalf("database was closed before runners drained: %v", err) + } +} + +func TestRunStartsEveryRunnerAndWaitsForAll(t *testing.T) { + db := newTestDB(t) + + const count = 5 + var started, finished atomic.Int32 + runners := make([]runner, 0, count) + for i := range count { + runners = append(runners, runner{ + name: "r", + run: func(ctx context.Context) { + started.Add(1) + <-ctx.Done() + // Stagger the exits so a missing Wait shows up reliably. + time.Sleep(time.Duration(i+1) * 20 * time.Millisecond) + finished.Add(1) + }, + }) + } + + srv := newStubServer(nil) + a := &App{ + db: db, + server: srv, + runners: runners, + addr: "127.0.0.1:0", + recoverInflight: func(context.Context) {}, + } + + ctx, cancel := context.WithCancel(context.Background()) + done := runInBackground(a, ctx) + srv.waitStarted(t) + cancel() + + if err := awaitRun(t, done); err != nil { + t.Fatalf("run returned an error: %v", err) + } + if got := started.Load(); got != count { + t.Errorf("started %d runners, want %d", got, count) + } + if got := finished.Load(); got != count { + t.Errorf("run returned with only %d of %d runners drained", got, count) + } +} + +// A server failure must tear the whole app down and surface as a non-nil +// error, so `openbee server` exits non-zero. +func TestRunReturnsServerErrorAndDrainsRunners(t *testing.T) { + db := newTestDB(t) + wantErr := errors.New("listen tcp: address already in use") + + var drained atomic.Bool + probe := runner{name: "probe", run: func(ctx context.Context) { + <-ctx.Done() + drained.Store(true) + }} + + srv := newStubServer(wantErr) + a := &App{ + db: db, + server: srv, + runners: []runner{probe}, + addr: "127.0.0.1:0", + recoverInflight: func(context.Context) {}, + } + + err := awaitRun(t, runInBackground(a, context.Background())) + if !errors.Is(err, wantErr) { + t.Fatalf("run error = %v, want it to wrap %v", err, wantErr) + } + if !drained.Load() { + t.Error("runners were not drained after the server failed") + } +} + +// Recovery must complete before any runner observes the world. +func TestRunRecoversBeforeStartingRunners(t *testing.T) { + db := newTestDB(t) + + var recovered, ranBeforeRecovery atomic.Bool + probe := runner{name: "probe", run: func(ctx context.Context) { + if !recovered.Load() { + ranBeforeRecovery.Store(true) + } + <-ctx.Done() + }} + + srv := newStubServer(nil) + a := &App{ + db: db, + server: srv, + runners: []runner{probe}, + addr: "127.0.0.1:0", + recoverInflight: func(context.Context) { + time.Sleep(50 * time.Millisecond) + recovered.Store(true) + }, + } + + ctx, cancel := context.WithCancel(context.Background()) + done := runInBackground(a, ctx) + srv.waitStarted(t) + cancel() + + if err := awaitRun(t, done); err != nil { + t.Fatalf("run returned an error: %v", err) + } + if !recovered.Load() { + t.Fatal("recovery never ran") + } + if ranBeforeRecovery.Load() { + t.Error("a runner started before startup recovery completed") + } +} diff --git a/internal/routes/server.go b/internal/routes/server.go index 3005c9a3..67619715 100644 --- a/internal/routes/server.go +++ b/internal/routes/server.go @@ -4,6 +4,7 @@ import ( "context" "io/fs" "net/http" + "sync" "github.com/gin-contrib/gzip" "github.com/gin-gonic/gin" @@ -35,8 +36,13 @@ type ServerParams struct { } type Server struct { - router *gin.Engine + router *gin.Engine + + // mu guards httpServer, which Run writes and Shutdown reads from a + // different goroutine. + mu sync.Mutex httpServer *http.Server + ServerParams } @@ -72,16 +78,22 @@ func (s *Server) setupRoutes() error { } func (s *Server) Run(addr string) error { - s.httpServer = &http.Server{ + srv := &http.Server{ Addr: addr, Handler: s.router, } - return s.httpServer.ListenAndServe() + s.mu.Lock() + s.httpServer = srv + s.mu.Unlock() + return srv.ListenAndServe() } func (s *Server) Shutdown(ctx context.Context) error { - if s.httpServer == nil { + s.mu.Lock() + srv := s.httpServer + s.mu.Unlock() + if srv == nil { return nil } - return s.httpServer.Shutdown(ctx) + return srv.Shutdown(ctx) }