// Command specsrht is the spec.sr.ht daemon, and — invoked under another name // — the receive hooks of every space it serves. // // It runs three things in one process: // // - the hook RPC socket at /.specsrht/hook.sock, which the pre-receive, // update and post-receive hooks of every space call. Validation lives here // and not in the hooks because bleve is single-writer and this process holds // the index, and because the push path and the API must not be able to // disagree about what is valid. // - an HTTP listener on -b (default localhost:5091), serving /healthz, the // read UI, /mcp for agents and /query for GraphQL — one listener, so one // reverse-proxy route covers the service. // - the reconciler, at startup and then periodically, repairing the // divergence a killed daemon leaves between git refs, Postgres and the // index. // // # Running as a hook // // Every hook a space's repository runs is a symlink to this binary. When // argv[0] names a hook the process handles that hook and exits, touching // neither the configuration nor the database; see the hooks package. The // explicit form `specsrht hook ` does the same thing by hand. // // # Admin commands // // Two subcommands run and exit without binding anything, so they are safe to // invoke while the daemon is up: // // specsrht space create ~owner/name | specsrht space list // specsrht doc propose ~owner/space ... [--as path] [--title t] // // The first is the only entry point a space has, and a deployment without one // is inert: there is nothing to hold documents. The second opens a proposal // from files on this host, for when the documents and the operator are already // here and a bearer token would be ceremony. // // There is no `specsrht token`. It minted spec's own agent credential; agent // credentials are tokens.sr.ht working tokens now, so they are minted there — // its /tokens page, or POST /exchange with a parent token — and carry the // spec:propose grant to write and spec:read to read. // // # Flags // // Parsed by core-go's server.New: // // -b addr bind address (repeatable); default localhost:5091 // -d debug: verbose request logging in core-go, and — read straight out // of the argument vector by sr-ht-ecore's logging.Defaults, before // server.New parses anything — debug verbosity for this daemon's own // logger from its first line. $LOG_LEVEL does the same for one run, // and [spec.sr.ht] log-level is the instance's persistent setting. // -m addr Prometheus metrics bind (default random port) // -p addr pprof bind (default random localhost port) // // Configuration comes from the shared SourceHut config.ini through core-go's // fixed search path. Every required key is checked before anything is opened, // and all missing ones are reported together, so a misconfigured instance // fails once at startup rather than once per restart. // // # Shutdown // // SIGINT and SIGTERM both start a warm shutdown. core-go's server.Run only // handles SIGINT — which is why compare.sr.ht's systemd unit sets // KillSignal=SIGINT — so this daemon installs a handler that turns a SIGTERM // into that same SIGINT. A unit for this service therefore needs no // KillSignal= line: the systemd default works. package main import ( "context" "database/sql" "errors" "fmt" "log/slog" "net/http" "os" "os/signal" "path/filepath" "strings" "syscall" "time" "github.com/99designs/gqlgen/graphql" "github.com/go-chi/chi/v5" chimw "github.com/go-chi/chi/v5/middleware" _ "github.com/lib/pq" // registers the "postgres" database/sql driver "github.com/vaughan0/go-ini" "go.bigb.es/auxilia/scribe" "sourcecraft.dev/bigbes/sr-ht-ecore/logging" "sourcecraft.dev/bigbes/sr-ht-core/auth" "sourcecraft.dev/bigbes/sr-ht-core/config" coreserver "sourcecraft.dev/bigbes/sr-ht-core/server" "sourcecraft.dev/bigbes/sr-ht-core/webhooks" "sourcecraft.dev/bigbes/sr-ht-spec/api" "sourcecraft.dev/bigbes/sr-ht-spec/core" "sourcecraft.dev/bigbes/sr-ht-spec/graph" "sourcecraft.dev/bigbes/sr-ht-spec/hooks" "sourcecraft.dev/bigbes/sr-ht-spec/mcpsrv" "sourcecraft.dev/bigbes/sr-ht-spec/search" "sourcecraft.dev/bigbes/sr-ht-spec/service" "sourcecraft.dev/bigbes/sr-ht-spec/web" ) const ( // serviceName is the SourceHut service identifier and our config section. // The ".sr.ht" suffix is what puts us in the nav network list. serviceName = "spec.sr.ht" // defaultBind is the address core-go binds when no -b is given. defaultBind = "localhost:5091" // version is reported in the MCP handshake so a client listing several // SourceHut MCP endpoints can tell which build it is talking to. version = "dev" // pingTimeout bounds the startup connectivity check against Postgres. pingTimeout = 10 * time.Second // shutdownGrace is how long the hook socket is given to finish the calls // already in flight once the HTTP listener has drained. A push being // validated at that moment finishes rather than being failed closed. shutdownGrace = 30 * time.Second ) func main() { // Hook mode first: a hook must not read a config file, open Postgres, or // bind anything. It talks to the daemon over a socket and exits. if _, _, isHook := hooks.ModeFromArgs(os.Args); isHook { os.Exit(hooks.Run(hooks.Runtime{Args: os.Args})) } // The config is read before the logger so that [spec.sr.ht] log-level is in // force for the first line this process writes. It cannot fail: core-go's // LoadConfig answers a nil ini.File for an instance with no config file at // all, and validateConfig — which runs later, in run — is what turns that // into one message an operator can act on. conf := config.LoadConfig() log := installLogger(conf) // Admin subcommands run and exit without binding anything, so they are safe // to invoke while the daemon holds the hook socket. if len(os.Args) > 1 { admin := map[string]func([]string) error{ "space": runSpace, "doc": runDoc, } if cmd := os.Args[1]; admin[cmd] != nil { if err := admin[cmd](os.Args[2:]); err != nil { fmt.Fprintf(os.Stderr, "specsrht %s: %v\n", cmd, err) os.Exit(1) } return } } if err := run(conf, log); err != nil { // Plain text, not a log record. A startup failure is read by a human // on a terminal, and the configuration report is deliberately several // lines long — a structured handler would escape it into one. fmt.Fprintf(os.Stderr, "spec.sr.ht did not start: %v\n", err) os.Exit(1) } } // runSpace is the space administration command: // // specsrht space create ~owner/name // specsrht space list // // Spaces have no other entry point. The read plane only reads, and the write // plane is the proposal API, which operates on documents inside a space that // already exists — so without this, a freshly deployed instance has no way to // hold anything at all. // // It installs the receive hooks itself rather than leaving them to the daemon's // startup refresh: a space created while the daemon is running would otherwise // accept unvalidated pushes until the next restart, which is exactly the // fail-open the receive path exists to prevent. func runSpace(args []string) error { if len(args) == 0 { return errors.New("usage: specsrht space create ~owner/name | specsrht space list") } conf := config.LoadConfig() cfg, err := validateConfig(conf) if err != nil { return err } pool, err := openDatabase(cfg.ConnectionString) if err != nil { return err } defer pool.Close() svc, err := service.New(cfg, pool) if err != nil { return err } ctx := context.Background() switch args[0] { case "create": if len(args) != 2 { return errors.New("usage: specsrht space create ~owner/name") } ref, err := core.ParseSpaceRef(args[1]) if err != nil { return fmt.Errorf("parse %q: %w", args[1], err) } if _, err := svc.CreateSpace(ctx, ref); err != nil { return err } binary, err := os.Executable() if err != nil { return fmt.Errorf("locate this binary, which every hook symlinks to: %w", err) } if err := hooks.InstallSpace(cfg.Repos, ref, hooks.InstallOptions{Binary: binary}); err != nil { return fmt.Errorf("install receive hooks for %s: %w", ref, err) } fmt.Printf("created %s\n repo: %s\n clone: git clone %s\n", ref, filepath.Join(cfg.Repos, "~"+ref.Owner, ref.Name), filepath.Join(cfg.Repos, "~"+ref.Owner, ref.Name)) return nil case "list": spaces, err := svc.ListSpaces(ctx) if err != nil { return err } for _, sp := range spaces { fmt.Println(sp.Ref) } return nil default: return fmt.Errorf("unknown subcommand %q: want create or list", args[0]) } } // installLogger builds the process logger and makes it slog's default. // // The policy — the verbosity, the source positions, the colour decision and the // set of attribute keys that must never reach a log file — is sr-ht-ecore's // logging.Defaults, because none of it is spec.sr.ht's to decide. -d, $LOG_LEVEL // and [spec.sr.ht] log-level are how an operator addresses every daemon on this // instance, and what must be redacted (the unified-login cookie, tokens.sr.ht // working tokens, the Authorization header they travel in) is a fact about the // instance rather than about this service. The local mask list is gone; the // network-key and private-key entries it contributed are in the shared one, and // so are the dsn and connection-string names it did not have. // // The handler stays here, and it is scribe's, because ecore deliberately builds // none. Everything goes to stderr: a hook's stdout is forwarded to the pushing // client, and this binary is both programs. // // logging.Install is what makes the library packages loggable at all. The read // plane, the credential resolver and sr-ht-ecore's panic middleware all log // through the default logger and none of them takes a *slog.Logger — a // middleware in another module cannot be handed this one, and without the // SetDefault its panic reports would come out of Go's plain stderr handler with // none of the masking applied. func installLogger(conf ini.File) *slog.Logger { opts := logging.Defaults(conf, serviceName) return logging.Install(scribe.NewTintHandler( scribe.WithWriter(os.Stderr), scribe.WithLevel(opts.Level), scribe.WithSource(opts.AddSource), scribe.WithTimeFormat(opts.TimeFormat), scribe.WithNoColor(!opts.Color), scribe.WithMaskKeys(opts.MaskKeys...), scribe.WithMask(opts.MaskPattern, opts.MaskReplacement), )) } func run(conf ini.File, log *slog.Logger) error { // The config was read in main, before the logger, so that a verbosity written // in config.ini is in force for the first line this process writes. It never // fails on a missing file — core-go's LoadConfig returns a nil ini.File — so // validateConfig is what turns an unconfigured instance into one clear // message instead of a panic deep inside the first request. cfg, err := validateConfig(conf) if err != nil { return err } pool, err := openDatabase(cfg.ConnectionString) if err != nil { return err } defer pool.Close() // WithInstanceTokens builds the daemon's one agent credential plane. It is // required, not offered: spec.sr.ht mints no credential of its own any more, // so an instance with no [tokens.sr.ht] section could authenticate no agent // at all, over HTTP or over `git push`. service.New fails here rather than // letting the daemon come up and refuse every agent one request at a time. svc, err := service.New(cfg, pool, service.WithInstanceTokens(conf)) if err != nil { return err } log.Info("agent credential plane", "tokens.sr.ht", svc.Resolver().HasInstancePlane()) // Seed the owner's user row before serving. core-go's auth.Middleware looks // a request's username up in the "user" table and, on a miss, calls out to // meta.sr.ht — seeding the single owner up front keeps that lookup local, and // it gives the webhook engine the user_id it scopes subscriptions by. if err := svc.EnsureOwnerUser(context.Background()); err != nil { return fmt.Errorf("seed the owner user row: %w", err) } // Refresh every space's hooks before anything can be pushed to it. This is // fatal on failure by design: a space whose hooks are missing accepts // pushes that are never validated, which is the one outcome the whole // receive path exists to prevent. A daemon that will not start is loud; a // space quietly accepting malformed documents is not. binary, err := os.Executable() if err != nil { return fmt.Errorf("locate this binary, which every hook symlinks to: %w", err) } if err := refreshHooks(context.Background(), log, svc, binary); err != nil { return err } surf, err := newSurfaces(conf, cfg, svc, version) if err != nil { return err } defer surf.Close() // After the surfaces, because the push notifier reindexes through the same // index they read from — one bleve writer, held here. hookSrv, err := hooks.NewServer(hooks.Options{ Backend: svc, Socket: hooks.SocketPath(cfg.Repos), Log: log, OnPush: pushNotifier(log, svc, surf.index), }) if err != nil { return err } if err := hookSrv.Listen(); err != nil { return err } // server.New parses -b/-d/-m/-p and runs crypto.InitCrypto(conf), whose // two required keys validateConfig already checked, so it cannot fatal // here for a reason we have not already reported. // /query is served on the authenticated router so core-go installs the // auth/database/server context the webhook engine needs — WithDefaultMiddleware // brings that whole stack (and, deliberately, core-go auth: agents therefore // read via MCP/REST, not /query). ownerOnly restricts /query to the instance // owner and maps them to AUTH_INTERNAL so the webhook engine's // NewAuthConfig/FilterWebhooks (which refuse cookie auth) accept them. // WithQueues starts the webhook delivery worker with a context carrying that // same stack; the queue executes a subscription's stored query against the // shared schema at delivery time. // The scope list must be an empty slice and not nil. core-go serves it // verbatim at /query/api-meta.json, where a nil slice marshals to // `"scopes": null` — and meta.sr.ht's OAuth page iterates that field for // every service it discovers, so one null there is a 500 on // /oauth2/personal-token for the whole instance, not a degraded entry. // Empty is also the honest answer: this service is owner-only (see // ownerOnly above) and defines no AccessScope enum to grant against. webhookQueue := webhooks.NewQueue(surf.schema, conf) srv := coreserver.New(serviceName, defaultBind, conf, os.Args). WithDefaultMiddleware(). WithMiddleware(ownerOnly(cfg.Instance.OwnerName)). WithSchema(surf.schema, []string{}). WithQueues(webhookQueue.Queue) mountRoutes(srv.AnonRouter(), conf, surf) // Now that the webhook queue is started (WithQueues gave its worker the // server+database+config context), install the sink so proposal lifecycle // events fire deliveries. The owner user id is valid — EnsureOwnerUser ran // above. svc.SetEventSink(newWebhookEventSink(webhookQueue, svc.OwnerUserID(), cfg.Instance.OwnerName, log)) ctx, stop := context.WithCancel(context.Background()) defer stop() served := make(chan error, 1) go func() { served <- hookSrv.Serve(ctx) }() go svc.RunReconciler(ctx, service.DefaultReconcileInterval, reconcileReporter(log)) bridgeSIGTERM(log) log.Info("spec.sr.ht starting", "bind", defaultBind, "repos", cfg.Repos, "cache", cfg.Cache, "origin", cfg.Origin, "hook_socket", hookSrv.Socket(), "reconcile_interval", service.DefaultReconcileInterval.String(), ) // Blocks until SIGINT — which bridgeSIGTERM makes SIGTERM equivalent to — // and then drains the HTTP listeners. srv.Run() log.Info("draining the hook socket", "grace", shutdownGrace.String()) stop() select { case err := <-served: if err != nil { log.Error("hook socket stopped with an error", scribe.Err(err)) } case <-time.After(shutdownGrace): log.Warn("hook socket did not drain in time; closing it") } if err := hookSrv.Close(); err != nil { log.Error("could not close the hook socket", scribe.Err(err)) } log.Info("spec.sr.ht stopped") return nil } // validateConfig checks every key this daemon needs before anything is opened, // and reports all of the missing ones at once so an operator fixes the config // in one pass instead of discovering each gap on a separate restart. // // service.LoadConfig owns our own section and collects its own gaps the same // way; the two lists are merged into one message. The keys checked here are // the ones core-go itself fatals on, which belong to server.New's contract // rather than to service/ — duplicating them there would give the instance two // lists to keep in sync. func validateConfig(conf ini.File) (service.Config, error) { var missing []string require := func(section, key, why string) { if v, ok := conf.Get(section, key); !ok || strings.TrimSpace(v) == "" { missing = append(missing, fmt.Sprintf("[%s] %s — %s", section, key, why)) } } // Both are read by crypto.InitCrypto, which server.New calls and which // fatals with a terse message when either is absent. The webhook key is // required even though v1 emits no webhooks. require("sr.ht", "network-key", "fernet key for the unified-login cookie") require("webhooks", "private-key", "webhook signing key; crypto.InitCrypto requires it") cfg, cfgErr := service.LoadConfig(conf) if len(missing) == 0 && cfgErr == nil { return cfg, nil } var b strings.Builder b.WriteString("incomplete configuration.") if len(missing) > 0 { fmt.Fprintf(&b, "\n\nMissing keys the SourceHut runtime requires:\n\t%s", strings.Join(missing, "\n\t")) } if cfgErr != nil { fmt.Fprintf(&b, "\n\n%s", cfgErr) } return service.Config{}, errors.New(b.String()) } // openDatabase opens the pool the whole daemon shares — request handlers, the // reconciler and the hook RPC alike — and proves it works before serving. // // sql.Open alone connects lazily, so a wrong DSN would first surface as a // rejected push. Fail-closed makes that safe but not pleasant; failing at // startup names the problem while somebody is still watching. func openDatabase(dsn string) (*sql.DB, error) { pool, err := sql.Open("postgres", dsn) if err != nil { return nil, fmt.Errorf("open the database: %w", err) } ctx, cancel := context.WithTimeout(context.Background(), pingTimeout) defer cancel() if err := pool.PingContext(ctx); err != nil { pool.Close() return nil, fmt.Errorf("reach the database: %w", err) } return pool, nil } // refreshHooks installs this binary's hooks into every space. // // It runs at startup rather than only at space creation so that an upgrade // which changes the wire protocol, the socket path or the hook set repairs // every repository by restarting — there is no separate migration step and no // repository left speaking last week's protocol. func refreshHooks(ctx context.Context, log *slog.Logger, svc *service.Service, binary string) error { spaces, err := svc.ListSpaces(ctx) if err != nil { return fmt.Errorf("list spaces to refresh their hooks: %w", err) } for _, sp := range spaces { if err := hooks.InstallSpace(svc.ReposRoot(), sp.Ref, hooks.InstallOptions{Binary: binary}); err != nil { return fmt.Errorf("install the receive hooks of %s: %w "+ "(a space whose hooks are missing would accept unvalidated pushes, so this is fatal; "+ "repair or remove the repository and start again)", sp.Ref, err) } } log.Info("receive hooks refreshed", "spaces", len(spaces), "binary", binary) return nil } // mountRoutes installs what the daemon serves over HTTP. func mountRoutes(router chi.Router, conf ini.File, surfaces *surfaces) { // server.New already froze the anonymous router for direct middleware // registration, so middleware and routes go in together inside a Group — // which chi permits on a fresh inline mux sharing the same routing tree. router.Group(func(r chi.Router) { r.Use(chimw.RealIP) r.Use(chimw.Recoverer) r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "text/plain; charset=utf-8") fmt.Fprint(w, "ok") }) }) mountWeb(router, conf, surfaces) } // surfaces are the three Phase 2 read surfaces, assembled once at startup. // // They share one *search.Index deliberately: bleve is single-writer, so a second // Open on the same directory is not merely wasteful but wrong. type surfaces struct { index *search.Index web *web.Server mcp http.Handler api http.Handler // schema is the GraphQL executable schema. Unlike the other surfaces it is // not a mounted handler: /query is served by core-go's server (WithSchema) // on the authenticated router, because the webhook engine needs core-go's // auth/database/server context there. The same schema is also handed to the // webhook queue, which executes a subscription's stored query against it at // delivery time — one schema, both callers. schema graphql.ExecutableSchema } // newSurfaces opens the index and builds the three read surfaces over it. // // All three go through service/ and none of them re-derives addressing, the // read contract or the project filter — that shared layer is the whole reason // "read SPEC-0007" cannot mean three different things depending on which door // you knock on. func newSurfaces(conf ini.File, cfg service.Config, svc *service.Service, version string) (*surfaces, error) { indexPath := filepath.Join(cfg.Cache, "index") // A crashed rebuild leaves working directories beside the index; clearing // them before Open is always safe, since the index is a pure cache. if err := search.CleanStale(indexPath); err != nil { return nil, fmt.Errorf("clean stale index working dirs: %w", err) } index, err := search.Open(indexPath) if err != nil { return nil, fmt.Errorf("open search index at %s: %w", indexPath, err) } site, err := web.New(web.Options{ Conf: conf, Reader: web.NewReader(svc), Searcher: index, Resolver: svc.Resolver(), }) if err != nil { index.Close() return nil, fmt.Errorf("assemble the web UI: %w", err) } // The origin is the /mcp Host allowlist: the MCP SDK's own DNS-rebinding // guard cannot tell a reverse proxy from an attacker (both reach a loopback // listener with a non-loopback Host), so it is replaced by a check against // this value. Traefik must pass the Host header through or every call 403s. mcp, err := mcpsrv.Handler(mcpsrv.Backend{Docs: svc, Index: index, Write: svc}, version, cfg.Origin) if err != nil { index.Close() return nil, fmt.Errorf("assemble the MCP surface: %w", err) } // spec_propose resolves the acting agent from the bearer token on the tool // call, so /mcp needs the principal middleware the read tools never did. // It sets an anonymous principal when there is no token, which service.Propose // refuses — the ACL stays in service/, this only populates the identity. // // mcpsrv.Gate sits inside that middleware and closes the read plane: the read // tools (spec_search/spec_read/spec_list) enforced nothing on their own, so it // applies the owner+agents ACL to the whole surface — the same one graph's // /query and the web UI apply. spec_propose stays fail-closed in service/ too; // the gate just makes the read tools match. mcp = svc.Resolver().Middleware()(mcpsrv.Gate(mcp)) schema, err := graph.NewSchema(graph.Options{ Reader: svc, Searcher: index, Proposals: graph.NewProposals(svc), Resolver: svc.Resolver(), }) if err != nil { index.Close() return nil, fmt.Errorf("assemble the GraphQL schema: %w", err) } rest, err := api.New(api.Options{Writer: svc, Resolver: svc.Resolver()}) if err != nil { index.Close() return nil, fmt.Errorf("assemble the REST write surface: %w", err) } return &surfaces{index: index, web: site, mcp: mcp, api: rest.Handler(), schema: schema}, nil } func (s *surfaces) Close() error { if s == nil || s.index == nil { return nil } return s.index.Close() } // mountWeb attaches the anonymous-router HTTP surfaces: the web UI, the MCP // endpoint, and the REST write plane. /query is NOT here — it is served by // core-go's server on the authenticated router (see run), because the webhook // engine needs core-go's auth/database/server context, which only // WithDefaultMiddleware installs. // // Order is load-bearing: /mcp and /api are registered before the web UI, which // mounts at "/" and would otherwise swallow them as document paths — "mcp" and // "api" are both legal space names as far as the router is concerned. // // These three keep spec's own authentication (agent bearer tokens, // anonymous-capable reads, login redirects), which is why they stay on the anon // router rather than behind core-go's 401-by-default auth. // ownerOnly is the /query access gate and the webhook engine's auth adapter, in // one middleware. It runs after core-go's auth.Middleware (which has already // 401'd anyone without a valid credential and resolved a username), so: // // - A non-owner authenticated user is refused with 403. core-go's auth admits // any valid meta user — it JIT-creates a row on a table miss — but spec.sr.ht // is single-owner: only the configured owner-name may reach /query at all. // This restores what spec's own authn did (everyone but the owner is nobody) // now that /query is behind core-go auth. // - The owner's context is remapped to AUTH_INTERNAL. The webhook engine's // NewAuthConfig and FilterWebhooks refuse AUTH_COOKIE outright, and INTERNAL // bypasses the (unused) @access scope checks — so this remap is what lets the // single owner, authenticated by a web cookie, manage webhooks. func ownerOnly(owner string) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ac := auth.ForContext(r.Context()) if ac.Username != owner { http.Error(w, "spec.sr.ht: only the instance owner may use /query", http.StatusForbidden) return } internal := *ac internal.AuthMethod = auth.AUTH_INTERNAL next.ServeHTTP(w, r.WithContext(auth.Context(r.Context(), &internal))) }) } } func mountWeb(router chi.Router, _ ini.File, s *surfaces) { if s == nil { return } router.Handle("/mcp", s.mcp) router.Mount("/api", s.api) router.Mount("/", s.web.Handler()) } // pushNotifier is what the daemon does when a push lands. // // Phase 1 records it and nothing more. Reindexing and advancing the space's // index rev stamp are Phase 2's, because bleve and the stamp arrive together: // moving the stamp now, with no index behind it, would assert that the index // is current and remove the reconciler's only way of noticing that it is not. func pushNotifier(log *slog.Logger, svc *service.Service, index *search.Index) hooks.PushNotifier { return func(ctx context.Context, space core.SpaceRef, updates []hooks.RefUpdate) error { refs := make([]string, 0, len(updates)) for _, u := range updates { refs = append(refs, u.String()) } // A push that touches only proposal branches changes nothing the index // holds: the index carries the approved revision, and proposal content // is deliberately not searchable — surfacing unreviewed text in search // is the same leak as serving it from the read plane. approved := false for _, u := range updates { if !strings.HasPrefix(u.Ref, "refs/heads/"+core.ProposalPrefix) { approved = true break } } if !approved { log.Info("push landed; no reindex needed", "space", space.String(), "refs", refs) return nil } sp, err := svc.OpenSpace(ctx, space) if err != nil { return fmt.Errorf("open %s to reindex: %w", space, err) } rev, err := svc.ResolveRev(ctx, sp, service.ApprovedRev) if err != nil { return fmt.Errorf("resolve the approved head of %s: %w", space, err) } arc, bodies, err := svc.Archive(ctx, sp, service.ApprovedRev) if err != nil { return fmt.Errorf("read %s at %s: %w", space, rev, err) } docs, err := search.Extract(arc, bodies) if err != nil { return fmt.Errorf("project %s for indexing: %w", space, err) } stats, err := index.RebuildSpace(ctx, space, docs) if err != nil { return fmt.Errorf("reindex %s: %w", space, err) } // The stamp goes last and only on success. Written earlier it would // assert the index reflects a revision it does not, which is precisely // the staleness the reconciler exists to detect — and it would detect // nothing. if _, err := svc.Store().SetIndexStamp(ctx, sp.ID, rev); err != nil { return fmt.Errorf("stamp the index for %s at %s: %w", space, rev, err) } log.Info("push landed; space reindexed", "space", space.String(), "refs", refs, "rev", rev, "indexed", stats.Indexed, "deleted", stats.Deleted, "took", stats.Took.String()) return nil } } // reconcileReporter logs the outcome of each reconciler pass. A failure is not // fatal: the next pass tries again, and a daemon that exits on a transient // Postgres error takes the push path down with it. func reconcileReporter(log *slog.Logger) func(*service.ReconcileReport, error) { return func(rep *service.ReconcileReport, err error) { if err != nil { log.Error("reconciler pass failed", scribe.Err(err)) return } attrs := []any{ "spaces", rep.Spaces, "repaired", len(rep.Repaired), "stale_indexes", len(rep.Reindex), "failures", len(rep.Failures), } for _, r := range rep.Repaired { log.Info("reconciler repaired divergence", "repair", fmt.Sprint(r)) } for _, f := range rep.Failures { log.Warn("reconciler could not repair", "failure", fmt.Sprint(f)) } log.Info("reconciler pass complete", attrs...) } } // bridgeSIGTERM makes systemd's default stop signal work. // // core-go's server.Run listens for SIGINT only, which is why compare.sr.ht's // unit carries KillSignal=SIGINT. Rather than require that line here, a // SIGTERM is turned into the SIGINT server.Run is waiting for. Both signals // are registered before Run installs its own handler, so a signal arriving in // the gap is caught rather than killing the process outright. // // After the first shutdown signal server.Run calls signal.Reset(os.Interrupt), // restoring the default disposition — so a second signal terminates // immediately, which is the documented behaviour and is why this keeps // forwarding rather than stopping after one. func bridgeSIGTERM(log *slog.Logger) { sig := make(chan os.Signal, 2) signal.Notify(sig, syscall.SIGTERM, os.Interrupt) go func() { for s := range sig { if s != syscall.SIGTERM { continue } log.Info("SIGTERM received; starting the warm shutdown core-go waits for SIGINT to begin") if err := syscall.Kill(os.Getpid(), syscall.SIGINT); err != nil { log.Error("could not raise SIGINT for the warm shutdown", scribe.Err(err)) } } }() }