vole
Clone
Read only, no credentials:
https://rodent.software/vole.git
Write over SSH with a registered key:
git clone ssh://git@rodent.software:2222/vole.git
A free and open-source forge that lives in your Git repo.
| 1 | // Package server wires the components together and owns the listeners. |
| 2 | package server |
| 3 | |
| 4 | import ( |
| 5 | "context" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log/slog" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "time" |
| 14 | |
| 15 | "rodent.software/vole/ci/runs" |
| 16 | "rodent.software/vole/core/lfs" |
| 17 | "rodent.software/vole/forge" |
| 18 | "rodent.software/vole/forge/config" |
| 19 | "rodent.software/vole/forge/identity" |
| 20 | "rodent.software/vole/forge/index" |
| 21 | "rodent.software/vole/forge/instancerepo" |
| 22 | "rodent.software/vole/forge/pageview" |
| 23 | "rodent.software/vole/forge/repo" |
| 24 | "rodent.software/vole/forge/secret" |
| 25 | "rodent.software/vole/forge/session" |
| 26 | "rodent.software/vole/internal/compress" |
| 27 | "rodent.software/vole/server/githttp" |
| 28 | "rodent.software/vole/server/pages" |
| 29 | "rodent.software/vole/server/ssh" |
| 30 | "rodent.software/vole/server/web" |
| 31 | ) |
| 32 | |
| 33 | type Server struct { |
| 34 | ctx context.Context |
| 35 | cancel context.CancelFunc |
| 36 | Cfg *config.Config |
| 37 | Repos *repo.Store |
| 38 | Identity *identity.Store |
| 39 | Sessions *session.Store |
| 40 | Runs *runs.Store |
| 41 | Index *index.Store |
| 42 | LFS *lfs.Store |
| 43 | Locks *lfs.LockStore |
| 44 | PageViews *pageview.Store |
| 45 | // InstanceRepo is the repository whose tree customizes the instance. |
| 46 | InstanceRepo *instancerepo.Store |
| 47 | // Secrets is the per-repository store behind the settings page and the |
| 48 | // `vole secret` verbs. Values never leave it except to a caller it has audited. |
| 49 | Secrets *secret.Store |
| 50 | Pages *pages.Server |
| 51 | Proxy *githttp.Proxy |
| 52 | SSH *ssh.Server |
| 53 | Web *web.Web |
| 54 | Log *slog.Logger |
| 55 | mux *http.ServeMux |
| 56 | acme *acmeSetup |
| 57 | // metadataWake nudges the metadata event loop after a browser write. Pushes |
| 58 | // deliver the same information through the spool; this is the path a write made |
| 59 | // without a push uses. |
| 60 | metadataWake chan struct{} |
| 61 | } |
| 62 | |
| 63 | func New(cfg *config.Config, binPath string, log *slog.Logger) (*Server, error) { |
| 64 | if log == nil { |
| 65 | log = slog.Default() |
| 66 | } |
| 67 | // Refuse an unsafe configuration before anything is opened: a listener that |
| 68 | // must not be reachable, a development login next to a public URL, a size that |
| 69 | // cannot be satisfied. A warning here would be read by nobody. |
| 70 | if err := cfg.Validate(); err != nil { |
| 71 | return nil, err |
| 72 | } |
| 73 | inst, err := forge.Open(cfg, binPath) |
| 74 | if err != nil { |
| 75 | return nil, err |
| 76 | } |
| 77 | // A secret that is group- or world-readable is a secret already disclosed. |
| 78 | // Tighten what an older install left, say so, and refuse to run when the mode |
| 79 | // cannot be changed: an instance that cannot protect its keys must not serve. |
| 80 | if tightened, err := forge.TightenSecrets(cfg); err != nil { |
| 81 | inst.Close() |
| 82 | return nil, err |
| 83 | } else if len(tightened) > 0 { |
| 84 | log.Warn("tightened permissions on secret files", "paths", strings.Join(tightened, ", ")) |
| 85 | } |
| 86 | if inst.ImportedACLs > 0 { |
| 87 | log.Info("access rules imported", "repos", inst.ImportedACLs) |
| 88 | } |
| 89 | if n, err := inst.Repos.InstallAllHooks(); err != nil { |
| 90 | log.Warn("install hooks", "err", err) |
| 91 | } else if n > 0 { |
| 92 | log.Info("hooks installed", "repos", n) |
| 93 | } |
| 94 | acme, err := newACME(cfg, inst.Repos.Public) |
| 95 | if err != nil { |
| 96 | inst.Close() |
| 97 | return nil, err |
| 98 | } |
| 99 | if acme != nil { |
| 100 | log.Info("automatic HTTPS enabled", "addr", cfg.HTTPSAddr, "hosts", strings.Join(acme.hostNames(), ", "), "cache", cfg.ACMECachePath()) |
| 101 | if cfg.PagesDomain != "" { |
| 102 | log.Info("site certificates issued on demand", "domain", cfg.PagesDomain) |
| 103 | } |
| 104 | } |
| 105 | proxy := githttp.New(githttp.Config{ |
| 106 | Root: cfg.ReposDir(), |
| 107 | Open: inst.Repos.Open, |
| 108 | Log: log, |
| 109 | MaxConcurrent: cfg.GitHTTPMaxConcurrent, |
| 110 | MaxPerClient: cfg.GitHTTPMaxPerClient, |
| 111 | ClientIPHeader: cfg.GitHTTPClientIPHeader, |
| 112 | RequestTimeout: time.Duration(cfg.GitHTTPTimeoutSeconds) * time.Second, |
| 113 | }) |
| 114 | ssh, err := ssh.New(ssh.Config{ |
| 115 | Identity: inst.Identity, |
| 116 | Repos: inst.Repos, |
| 117 | Instance: inst, |
| 118 | Logs: inst.Logs, |
| 119 | LFS: inst.LFS, |
| 120 | LFSWarnBytes: cfg.LFSWarnBytes, |
| 121 | Log: log, |
| 122 | HostKeyPath: cfg.SSHHostKey(), |
| 123 | MaxBlobBytes: cfg.MaxBlobBytes, |
| 124 | MaxRefs: cfg.RepoMaxRefs, |
| 125 | MaxMetadataEntries: cfg.RepoMaxMetadataEntries, |
| 126 | MaxMetadataBytes: cfg.RepoMaxMetadataBytes, |
| 127 | MaxRepoBytes: cfg.RepoMaxBytes, |
| 128 | MaxLFSBytes: cfg.RepoMaxLFSBytes, |
| 129 | AuditPath: cfg.AuditLog(), |
| 130 | SpoolDir: cfg.SpoolDir(), |
| 131 | RunsDB: cfg.RunsDB(), |
| 132 | IndexDB: cfg.IndexDB(), |
| 133 | Runs: inst.Runs, |
| 134 | Locks: inst.Locks, |
| 135 | Lease: cfg.Lease(), |
| 136 | MaxJobsPerRepo: cfg.MaxJobsPerRepo, |
| 137 | PagesRef: cfg.PagesRef, |
| 138 | PublicURL: cfg.BaseURL(), |
| 139 | MaxConnections: cfg.SSHMaxConnections, |
| 140 | MaxGitProcesses: cfg.SSHMaxGitProcesses, |
| 141 | }) |
| 142 | if err != nil { |
| 143 | inst.Close() |
| 144 | return nil, fmt.Errorf("ssh: %w", err) |
| 145 | } |
| 146 | wp, err := web.New(inst.Repos, inst.Index, inst.Identity, inst.Runs, cfg) |
| 147 | if err != nil { |
| 148 | inst.Close() |
| 149 | return nil, err |
| 150 | } |
| 151 | wp.LFS = inst.LFS |
| 152 | wp.Locks = inst.Locks |
| 153 | wp.Sessions = inst.Sessions |
| 154 | wp.PageViews = inst.PageViews |
| 155 | wp.InstanceRepo = inst.InstanceRepo |
| 156 | wp.Secrets = inst.Secrets |
| 157 | wp.Logger = func(format string, args ...any) { log.Warn(fmt.Sprintf(format, args...)) } |
| 158 | if cfg.DevLogin != "" { |
| 159 | log.Warn("development login enabled", "account", cfg.DevLogin, "http_addr", cfg.HTTPAddr) |
| 160 | } |
| 161 | if wp.Auth != nil && !wp.Auth.Enabled() { |
| 162 | log.Info("browser login disabled", "reason", wp.Auth.Disabled) |
| 163 | } else if cfg.OpenRegistration { |
| 164 | log.Info("open registration enabled", "url", cfg.BaseURL()+"/-/register") |
| 165 | } |
| 166 | |
| 167 | s := &Server{ |
| 168 | Cfg: cfg, Repos: inst.Repos, Identity: inst.Identity, Sessions: inst.Sessions, |
| 169 | Runs: inst.Runs, Index: inst.Index, acme: acme, |
| 170 | LFS: inst.LFS, Locks: inst.Locks, PageViews: inst.PageViews, Secrets: inst.Secrets, |
| 171 | Proxy: proxy, SSH: ssh, Web: wp, Log: log, |
| 172 | metadataWake: make(chan struct{}, 1), |
| 173 | InstanceRepo: inst.InstanceRepo, |
| 174 | } |
| 175 | wp.Logs = inst.Logs |
| 176 | wp.AllowCI = s.allowContributorCI |
| 177 | wp.EnsureCandidate = s.ensureCandidateForWeb |
| 178 | wp.MetadataChanged = s.wakeMetadata |
| 179 | wp.DispatchWorkflow = s.dispatchWorkflow |
| 180 | ssh.SetDispatchWorkflow(s.dispatchWorkflow) |
| 181 | wp.RenameRepo = inst.RenameRepo |
| 182 | wp.DeleteRepo = inst.DeleteRepo |
| 183 | // Sites are opt-in: without a pages domain there is no host to serve them on, |
| 184 | // and an instance that never wanted them pays no processes for the idea. |
| 185 | if cfg.PagesDomain != "" { |
| 186 | s.Pages = pages.New(pages.Config{ |
| 187 | Domain: cfg.PagesDomain, |
| 188 | Ref: cfg.PagesRef, |
| 189 | Procs: cfg.PagesProcs, |
| 190 | CacheBytes: cfg.PagesCacheBytes, |
| 191 | Log: log, |
| 192 | }, inst.Repos) |
| 193 | } |
| 194 | s.mux = s.routes() |
| 195 | // Background loops start here rather than in Run: a caller that serves the |
| 196 | // listeners itself (the tests, an embedder) must still get roster reloads, |
| 197 | // queue ingestion and lease expiry. |
| 198 | s.ctx, s.cancel = context.WithCancel(context.Background()) |
| 199 | if s.InstanceRepo != nil && s.InstanceRepo.Enabled() { |
| 200 | if err := s.InstanceRepo.Reload(); err != nil { |
| 201 | log.Warn("instance repository: cannot load", "repo", cfg.InstanceRepo, "err", err) |
| 202 | } |
| 203 | go s.watchInstanceRepo(s.ctx) |
| 204 | } |
| 205 | go s.watchRoster(s.ctx) |
| 206 | go s.watchSpool(s.ctx) |
| 207 | go s.watchRefs(s.ctx) |
| 208 | go s.watchMetadata(s.ctx) |
| 209 | go s.watchSchedules(s.ctx) |
| 210 | go s.watchCandidates(s.ctx) |
| 211 | go s.watchAutoMerges(s.ctx) |
| 212 | go s.watchLeases(s.ctx) |
| 213 | // An instance that stored objects before they were scoped by repository needs its |
| 214 | // links, and nobody reads an upgrade note. This is idempotent and deletes nothing. |
| 215 | s.scopeLFSStore() |
| 216 | return s, nil |
| 217 | } |
| 218 | |
| 219 | // Handler returns the HTTP surface. Site hosts are dispatched first: a request |
| 220 | // for <repo>.<pages_domain> is a page, and everything else is the forge UI. |
| 221 | // |
| 222 | // Sites are compressed here rather than inside the pages package because the |
| 223 | // dispatcher is the last point both surfaces pass through, and a site is a static |
| 224 | // file server: HTML, CSS and JavaScript, which is exactly what gzip is for. The |
| 225 | // forge UI does its own, inside its middleware, because the compressor has to sit |
| 226 | // under the status recorder there. |
| 227 | func (s *Server) Handler() http.Handler { |
| 228 | forge := s.Web.Middleware(s.forgeHandler(s.mux)) |
| 229 | if s.Pages == nil { |
| 230 | return forge |
| 231 | } |
| 232 | sites := compress.Handler(s.Pages) |
| 233 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 234 | if _, ok := s.Pages.Host(r.Host); ok { |
| 235 | sites.ServeHTTP(w, r) |
| 236 | return |
| 237 | } |
| 238 | forge.ServeHTTP(w, r) |
| 239 | }) |
| 240 | } |
| 241 | |
| 242 | // drainGrace is how long a clean shutdown lets an already-leased job finish before |
| 243 | // closing the listeners. It is shorter than systemd's default stop timeout, which |
| 244 | // leaves room for the HTTP server's own shutdown. Work that does not fit in the |
| 245 | // window is still lost on a restart; a result that arrives inside it is not. |
| 246 | const drainGrace = time.Minute |
| 247 | |
| 248 | // drain stops new leases and waits for the jobs already running to report. It is |
| 249 | // best effort: a runner that died, or a build longer than the grace, is left for |
| 250 | // the next process, which can requeue it from the durable lease. |
| 251 | func (s *Server) drain() { |
| 252 | if s.Runs == nil { |
| 253 | return |
| 254 | } |
| 255 | s.Runs.Drain() |
| 256 | n, err := s.Runs.Running() |
| 257 | if err != nil { |
| 258 | s.Log.Warn("drain: cannot count running jobs", "err", err) |
| 259 | return |
| 260 | } |
| 261 | if n == 0 { |
| 262 | return |
| 263 | } |
| 264 | s.Log.Info("draining running jobs", "jobs", n, "grace", drainGrace) |
| 265 | deadline := time.Now().Add(drainGrace) |
| 266 | for time.Now().Before(deadline) { |
| 267 | time.Sleep(250 * time.Millisecond) |
| 268 | n, err = s.Runs.Running() |
| 269 | if err != nil { |
| 270 | s.Log.Warn("drain: cannot count running jobs", "err", err) |
| 271 | return |
| 272 | } |
| 273 | if n == 0 { |
| 274 | s.Log.Info("running jobs finished; shutting down") |
| 275 | return |
| 276 | } |
| 277 | } |
| 278 | s.Log.Warn("drain grace expired with jobs still running", "jobs", n) |
| 279 | } |
| 280 | |
| 281 | // Run starts the listeners and blocks until ctx is cancelled. With automatic |
| 282 | // HTTPS, the HTTP address only answers ACME challenges and redirects, and the |
| 283 | // forge is served over TLS on HTTPSAddr. |
| 284 | func (s *Server) Run(ctx context.Context) error { |
| 285 | sshLn, err := net.Listen("tcp", s.Cfg.SSHAddr) |
| 286 | if err != nil { |
| 287 | return fmt.Errorf("ssh listen %s: %w", s.Cfg.SSHAddr, err) |
| 288 | } |
| 289 | httpLn, err := net.Listen("tcp", s.Cfg.HTTPAddr) |
| 290 | if err != nil { |
| 291 | sshLn.Close() |
| 292 | return fmt.Errorf("http listen %s: %w", s.Cfg.HTTPAddr, err) |
| 293 | } |
| 294 | |
| 295 | httpSrv := &http.Server{ |
| 296 | Handler: s.Handler(), |
| 297 | ReadHeaderTimeout: 15 * time.Second, |
| 298 | // ReadTimeout bounds the request, body included. The only bodies on the |
| 299 | // HTTP surface are a login form, a comment and a fetch negotiation, all of |
| 300 | // which are small and arrive long before this. |
| 301 | ReadTimeout: 60 * time.Second, |
| 302 | // WriteTimeout is deliberately unset: a clone, a release tarball, an LFS |
| 303 | // object and a site file are streamed, and a wall-clock limit on writing the |
| 304 | // response would truncate a legitimate large download. Those streams are |
| 305 | // bounded where they are produced instead — the git transport by its own |
| 306 | // request deadline, archives by a concurrency bound, downloads by a size cap. |
| 307 | MaxHeaderBytes: 64 << 10, |
| 308 | IdleTimeout: 120 * time.Second, |
| 309 | } |
| 310 | var httpsLn net.Listener |
| 311 | var httpsSrv *http.Server |
| 312 | if s.acme != nil { |
| 313 | httpSrv.Handler = s.acme.httpHandler() |
| 314 | httpsLn, err = net.Listen("tcp", s.Cfg.HTTPSAddr) |
| 315 | if err != nil { |
| 316 | httpLn.Close() |
| 317 | sshLn.Close() |
| 318 | return fmt.Errorf("https listen %s: %w", s.Cfg.HTTPSAddr, err) |
| 319 | } |
| 320 | httpsSrv = &http.Server{ |
| 321 | Handler: s.Handler(), |
| 322 | TLSConfig: s.acme.tlsConfig(), |
| 323 | ReadHeaderTimeout: 15 * time.Second, |
| 324 | ReadTimeout: 60 * time.Second, |
| 325 | MaxHeaderBytes: 64 << 10, |
| 326 | IdleTimeout: 120 * time.Second, |
| 327 | } |
| 328 | } |
| 329 | |
| 330 | errCh := make(chan error, 3) |
| 331 | go func() { |
| 332 | if err := s.SSH.Serve(sshLn); err != nil { |
| 333 | errCh <- fmt.Errorf("ssh serve: %w", err) |
| 334 | } |
| 335 | }() |
| 336 | go func() { |
| 337 | if err := httpSrv.Serve(httpLn); err != nil && !errors.Is(err, http.ErrServerClosed) { |
| 338 | errCh <- fmt.Errorf("http serve: %w", err) |
| 339 | } |
| 340 | }() |
| 341 | if httpsSrv != nil { |
| 342 | go func() { |
| 343 | if err := httpsSrv.ServeTLS(httpsLn, "", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { |
| 344 | errCh <- fmt.Errorf("https serve: %w", err) |
| 345 | } |
| 346 | }() |
| 347 | } |
| 348 | |
| 349 | if httpsSrv != nil { |
| 350 | s.Log.Info("listening", "ssh", sshLn.Addr().String(), "http", httpLn.Addr().String(), |
| 351 | "https", httpsLn.Addr().String(), "data", s.Cfg.DataDir, "users", s.Identity.Snapshot().Count) |
| 352 | } else { |
| 353 | s.Log.Info("listening", "ssh", sshLn.Addr().String(), "http", httpLn.Addr().String(), |
| 354 | "data", s.Cfg.DataDir, "users", s.Identity.Snapshot().Count) |
| 355 | } |
| 356 | |
| 357 | select { |
| 358 | case <-ctx.Done(): |
| 359 | // Let the runners finish what they already hold before the sockets go away. |
| 360 | // A deploy that cuts a job in its last step costs a whole second attempt. |
| 361 | s.drain() |
| 362 | sh, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| 363 | defer cancel() |
| 364 | s.SSH.Close() |
| 365 | httpSrv.Shutdown(sh) |
| 366 | if httpsSrv != nil { |
| 367 | httpsSrv.Shutdown(sh) |
| 368 | } |
| 369 | return nil |
| 370 | case err := <-errCh: |
| 371 | s.SSH.Close() |
| 372 | httpSrv.Close() |
| 373 | if httpsSrv != nil { |
| 374 | httpsSrv.Close() |
| 375 | } |
| 376 | return err |
| 377 | } |
| 378 | } |
| 379 | |
| 380 | func (s *Server) watchRoster(ctx context.Context) { |
| 381 | t := time.NewTicker(5 * time.Second) |
| 382 | defer t.Stop() |
| 383 | for { |
| 384 | select { |
| 385 | case <-ctx.Done(): |
| 386 | return |
| 387 | case <-t.C: |
| 388 | before := s.Identity.Snapshot().Count |
| 389 | if err := s.Identity.Reload(); err != nil { |
| 390 | s.Log.Error("roster reload failed; keeping the previous snapshot", "err", err) |
| 391 | continue |
| 392 | } |
| 393 | if after := s.Identity.Snapshot().Count; after != before { |
| 394 | s.Log.Info("roster reloaded", "accounts", after) |
| 395 | } |
| 396 | } |
| 397 | } |
| 398 | } |
| 399 | |
| 400 | func (s *Server) Close() error { |
| 401 | if s.cancel != nil { |
| 402 | s.cancel() |
| 403 | } |
| 404 | if s.Pages != nil { |
| 405 | s.Pages.Close() |
| 406 | } |
| 407 | if s.Runs != nil { |
| 408 | _ = s.Runs.Close() |
| 409 | } |
| 410 | if s.Locks != nil { |
| 411 | _ = s.Locks.Close() |
| 412 | } |
| 413 | if s.PageViews != nil { |
| 414 | _ = s.PageViews.Close() |
| 415 | } |
| 416 | if s.Index != nil { |
| 417 | if err := s.Index.Close(); err != nil { |
| 418 | return err |
| 419 | } |
| 420 | } |
| 421 | if s.Sessions != nil { |
| 422 | _ = s.Sessions.Close() |
| 423 | } |
| 424 | if s.Identity != nil { |
| 425 | return s.Identity.Close() |
| 426 | } |
| 427 | return nil |
| 428 | } |
| 429 | |
| 430 | func (s *Server) RebuildIndex() (int, error) { return s.Index.Rebuild(s.Repos) } |
| 431 | |
| 432 | // MaxInputLabel is used by logging to show the effective push cap. |
| 433 | func (s *Server) MaxInputLabel() string { |
| 434 | return strconv.FormatInt(s.Cfg.ReceiveMaxBytes, 10) |
| 435 | } |
| 436 |