// 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 <repos>/.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 <name>` does the same thing by hand.
//
// # Admin commands
//
// Three 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 token create <name> | specsrht token list | specsrht token revoke <id>
// specsrht doc propose ~owner/space <file>... [--as path] [--title t]
//
// The first two are the only entry points spaces and agent tokens have. A
// deployment without both is inert: nothing to hold documents, and no credential
// for the remote agent write plane, which refuses an anonymous caller by design.
// The third 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.
//
// # Flags
//
// Parsed by core-go's server.New:
//
// -b addr bind address (repeatable); default localhost:5091
// -d debug (verbose request logging in core-go)
// -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-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}))
}
log := newLogger()
slog.SetDefault(log)
// 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,
"token": runToken,
"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(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])
}
}
// newLogger builds the process logger. LOG_LEVEL raises or lowers verbosity;
// everything goes to stderr, because a hook's stdout is forwarded to the
// pushing client and this binary is both programs.
func newLogger() *slog.Logger {
level := new(slog.LevelVar)
level.Set(parseLevel(os.Getenv("LOG_LEVEL")))
return slog.New(scribe.NewTintHandler(
scribe.WithWriter(os.Stderr),
scribe.WithLevel(level),
))
}
func parseLevel(s string) slog.Level {
switch strings.ToLower(strings.TrimSpace(s)) {
case "debug":
return slog.LevelDebug
case "warn", "warning":
return slog.LevelWarn
case "error":
return slog.LevelError
default:
return slog.LevelInfo
}
}
func run(log *slog.Logger) error {
// LoadConfig never fails on a missing file — it 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.
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()
// WithInstanceTokens offers the daemon's resolver the tokens.sr.ht bearer
// plane. It is offered, not required: an instance whose config has no
// [tokens.sr.ht] section gets no such plane and keeps accepting its own
// agent token, which is a supported configuration and not a degraded one.
svc, err := service.New(cfg, pool, service.WithInstanceTokens(conf))
if err != nil {
return err
}
log.Info("agent credential planes",
"local", true, "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.
webhookQueue := webhooks.NewQueue(surf.schema, conf)
srv := coreserver.New(serviceName, defaultBind, conf, os.Args).
WithDefaultMiddleware().
WithMiddleware(ownerOnly(cfg.Instance.OwnerName)).
WithSchema(surf.schema, nil).
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))
}
}
}()
}