pds.dad
This repository has no description
1package main
2
3import (
4 "context"
5 "fmt"
6 "log/slog"
7 "net/http"
8 "os"
9 "os/signal"
10 "syscall"
11 "time"
12
13 "github.com/bluesky-social/indigo/atproto/identity"
14 "github.com/earthboundkid/versioninfo/v2"
15 "github.com/prometheus/client_golang/prometheus/promhttp"
16 "github.com/urfave/cli/v3"
17 "golang.org/x/sync/errgroup"
18 "tangled.org/pds.dad/replica/internal"
19)
20
21func main() {
22 if err := run(os.Args); err != nil {
23 slog.Error("exiting process", "error", err)
24 os.Exit(-1)
25 }
26}
27
28func run(args []string) error {
29 ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
30 defer cancel()
31
32 app := &cli.Command{
33 Name: "replica",
34 Usage: "PDS repo replica tool",
35 Version: versioninfo.Short(),
36
37 Commands: []*cli.Command{
38 {
39 Name: "run",
40 Usage: "start the replica service",
41 Flags: []cli.Flag{
42 &cli.StringFlag{
43 Name: "pds",
44 Usage: "PDS URL to replicate",
45 Sources: cli.EnvVars("REPLICA_PDS"),
46 },
47 &cli.StringFlag{
48 Name: "db-url",
49 Usage: "sqlite database location",
50 Value: "./pds-replica.db",
51 Sources: cli.EnvVars("REPLICA_DATABASE_URL"),
52 },
53 &cli.StringFlag{
54 Name: "data-dir",
55 Usage: "directory to store repo CAR backups",
56 Value: "./data",
57 Sources: cli.EnvVars("REPLICA_DATA_DIR"),
58 },
59 &cli.DurationFlag{
60 Name: "scrub-interval",
61 Usage: "how often to verify on-disk CARs against the DB state",
62 Value: 24 * time.Hour,
63 Sources: cli.EnvVars("REPLICA_SCRUB_INTERVAL"),
64 },
65&cli.DurationFlag{
66 Name: "deleted-retention",
67 Usage: "how long to keep backups of deleted accounts before purging",
68 Value: 720 * time.Hour,
69 Sources: cli.EnvVars("REPLICA_DELETED_RETENTION"),
70 },
71 &cli.StringFlag{
72 Name: "metrics-listen",
73 Usage: "address for the Prometheus /metrics and pprof HTTP server (disabled if empty)",
74 Sources: cli.EnvVars("REPLICA_METRICS_LISTEN"),
75 },
76 },
77 Action: runReplica,
78 },
79 },
80 }
81 return app.Run(ctx, args)
82}
83
84func runReplica(ctx context.Context, cmd *cli.Command) error {
85 internal.SetupLogger()
86 pdsURL := cmd.String("pds")
87 if pdsURL == "" {
88 return cli.Exit("pds flag is required", 1)
89 }
90 dbURL := cmd.String("db-url")
91 dataDir := cmd.String("data-dir")
92
93 slog.Info("starting replica", "pds", pdsURL, "db", dbURL, "dataDir", dataDir)
94
95 db, err := internal.SetupDatabase(dbURL)
96 if err != nil {
97 return fmt.Errorf("failed to setup database: %w", err)
98 }
99
100 store, err := internal.NewCarStore(dataDir)
101 if err != nil {
102 return fmt.Errorf("failed to setup CAR store: %w", err)
103 }
104
105 // replay any root-patch journals left by a crash before anything reads
106 // or appends to the CARs
107 if err := store.RecoverAllJournals(); err != nil {
108 return fmt.Errorf("failed to recover root journals: %w", err)
109 }
110
111 logger := slog.Default()
112 idDir := identity.DefaultDirectory()
113 repoLocks := internal.NewKeyedMutex()
114
115 crawler := internal.NewCrawler(logger, db, pdsURL)
116 resyncer := internal.NewResyncer(logger, db, store, idDir, repoLocks, pdsURL, 4)
117 firehose := internal.NewFirehose(logger, db, store, idDir, repoLocks, pdsURL)
118 scrubber := internal.NewScrubber(logger, db, store, repoLocks, cmd.Duration("scrub-interval"), cmd.Duration("deleted-retention"))
119
120 g, ctx := errgroup.WithContext(ctx)
121 g.Go(func() error {
122 crawler.Run(ctx)
123 return nil
124 })
125 g.Go(func() error {
126 resyncer.Run(ctx)
127 return nil
128 })
129 g.Go(func() error {
130 return firehose.Run(ctx)
131 })
132 g.Go(func() error {
133 scrubber.Run(ctx)
134 return nil
135 })
136 // always run the gauges poller so per-state counts and disk bytes stay
137 // live for anyone scraping /metrics later
138 g.Go(func() error {
139 internal.RunMetricsCollector(ctx, db, dataDir)
140 return nil
141 })
142 if metricsAddr := cmd.String("metrics-listen"); metricsAddr != "" {
143 mux := http.NewServeMux()
144 mux.Handle("/metrics", promhttp.Handler())
145 mux.HandleFunc("/_health", func(w http.ResponseWriter, _ *http.Request) {
146 w.WriteHeader(http.StatusOK)
147 })
148 g.Go(func() error {
149 logger.Info("starting metrics server", "addr", metricsAddr)
150 srv := &http.Server{
151 Addr: metricsAddr,
152 Handler: mux,
153 ReadHeaderTimeout: 10 * time.Second,
154 }
155 go func() {
156 <-ctx.Done()
157 shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
158 defer cancel()
159 _ = srv.Shutdown(shutdownCtx)
160 }()
161 if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
162 return fmt.Errorf("metrics server failed: %w", err)
163 }
164 return nil
165 })
166 }
167
168 return g.Wait()
169}