brianmcgee
27ea5601c5
Introduces a `config.Now` field which all components should defer to when getting the current time instead of hardcoded calls to `model.DefaultClock`.
127 lines
3.4 KiB
Go
127 lines
3.4 KiB
Go
// Package server implements a daemon process for Data Mesher.
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"git.clan.lol/clan/data-mesher/pkg/cluster"
|
|
"git.clan.lol/clan/data-mesher/pkg/config"
|
|
"git.clan.lol/clan/data-mesher/pkg/http"
|
|
"git.clan.lol/clan/data-mesher/pkg/state"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
// Server represents the main application server, managing HTTP, clustering, logging, and stateful components.
|
|
type Server struct {
|
|
config *config.Config // app config
|
|
|
|
files *state.Files // file/signature store
|
|
http *http.Server // admin endpoints
|
|
cluster *cluster.Cluster // clustering via libp2p
|
|
|
|
eg *errgroup.Group // for background tasks
|
|
cancelTasks context.CancelFunc // for graceful shutdown of background tasks
|
|
}
|
|
|
|
// NewServer initializes a new Server instance using the provided configuration.
|
|
// Returns the configured Server or an error if initialization fails.
|
|
func NewServer(cfg *config.Config) (*Server, error) {
|
|
var err error
|
|
|
|
srv := &Server{
|
|
config: cfg,
|
|
eg: &errgroup.Group{},
|
|
}
|
|
|
|
// create file store
|
|
srv.files, err = state.NewFiles(cfg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create files store: %w", err)
|
|
}
|
|
|
|
// create a new cluster instance
|
|
if srv.cluster, err = cluster.New(cfg, srv.files, nil); err != nil {
|
|
return nil, fmt.Errorf("failed to create cluster: %w", err)
|
|
}
|
|
|
|
if srv.http, err = http.NewServer(cfg, srv.files, srv.cluster); err != nil {
|
|
return nil, fmt.Errorf("failed to create http server: %w", err)
|
|
}
|
|
|
|
return srv, nil
|
|
}
|
|
|
|
// Start initializes and starts the server, including background processes, HTTP server, and cluster membership
|
|
// handling.
|
|
// Returns an error if any initialization or background tasks fail.
|
|
func (s *Server) Start(ctx context.Context) error {
|
|
slog.Info("starting server")
|
|
|
|
// set up background processes
|
|
s.eg = &errgroup.Group{}
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
|
|
// capture cancel function for Stop()
|
|
s.cancelTasks = cancel
|
|
|
|
// start http server
|
|
s.eg.Go(func() error {
|
|
return s.http.Listen(ctx)
|
|
})
|
|
|
|
// join the cluster
|
|
if err := s.cluster.Join(ctx); err != nil {
|
|
return fmt.Errorf("failed to join cluster: %w", err)
|
|
}
|
|
|
|
// schedule background jobs
|
|
s.eg.Go(func() error {
|
|
return s.cluster.PushPullLoop(ctx)
|
|
})
|
|
|
|
// background sweeper for files whose ValidFor TTL has elapsed
|
|
s.eg.Go(func() error {
|
|
return s.files.SweepLoop(ctx, s.config.SweepInterval, s.config.Now)
|
|
})
|
|
|
|
slog.Info("server started")
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stop gracefully stops the server, shutting down background tasks, HTTP server, cluster, and store.
|
|
// If the provided context does not include a deadline, a default deadline of 30 seconds is used.
|
|
func (s *Server) Stop(ctx context.Context) error {
|
|
// get deadline for shutdown, extracting from context if there is one
|
|
deadline, ok := ctx.Deadline()
|
|
if !ok {
|
|
deadline = time.Now().Add(30 * time.Second)
|
|
}
|
|
|
|
// stop http server
|
|
if err := s.http.Stop(ctx); err != nil {
|
|
slog.Error("failed to stop http server", slog.Any("err", err))
|
|
}
|
|
|
|
// leave cluster
|
|
if err := s.cluster.Leave(ctx); err != nil {
|
|
slog.Error("failed to leave cluster within deadline", slog.Time("deadline", deadline), slog.Any("err", err))
|
|
}
|
|
|
|
// stop background tasks
|
|
s.cancelTasks()
|
|
|
|
// wait for background tasks to finish
|
|
if err := s.eg.Wait(); err != nil {
|
|
slog.Error("failure occurred waiting for background tasks", slog.Any("err", err))
|
|
}
|
|
|
|
slog.Info("server stopped")
|
|
|
|
return nil
|
|
}
|