pds.dad / replica
This repository has no description
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

metrics

author
Bailey Townsend date (Jul 19, 2026, 12:35 PM -0500) commit 1b53b4b2 parent 2785104e
+509 -61
+44 -6
cmd/main.go
··· 4 4 "context" 5 5 "fmt" 6 6 "log/slog" 7 + "net/http" 7 8 "os" 8 9 "os/signal" 9 10 "syscall" ··· 11 12 12 13 "github.com/bluesky-social/indigo/atproto/identity" 13 14 "github.com/earthboundkid/versioninfo/v2" 15 + "github.com/prometheus/client_golang/prometheus/promhttp" 14 16 "github.com/urfave/cli/v3" 15 17 "golang.org/x/sync/errgroup" 16 18 "tangled.org/pds.dad/replica/internal" ··· 60 62 Value: 24 * time.Hour, 61 63 Sources: cli.EnvVars("REPLICA_SCRUB_INTERVAL"), 62 64 }, 63 - &cli.DurationFlag{ 64 - Name: "deleted-retention", 65 - Usage: "how long to keep backups of deleted accounts before purging", 66 - Value: 720 * time.Hour, 67 - Sources: cli.EnvVars("REPLICA_DELETED_RETENTION"), 68 - }, 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 + }, 69 76 }, 70 77 Action: runReplica, 71 78 }, ··· 126 133 scrubber.Run(ctx) 127 134 return nil 128 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 + } 129 167 130 168 return g.Wait() 131 169 }
+1 -1
go.mod
··· 9 9 github.com/ipfs/go-cid v0.4.1 10 10 github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 11 11 github.com/multiformats/go-multihash v0.2.3 12 + github.com/prometheus/client_golang v1.17.0 12 13 github.com/spf13/viper v1.21.0 13 14 github.com/urfave/cli/v3 v3.10.1 14 15 golang.org/x/sync v0.22.0 ··· 65 66 github.com/opentracing/opentracing-go v1.2.0 // indirect 66 67 github.com/pelletier/go-toml/v2 v2.2.4 // indirect 67 68 github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect 68 - github.com/prometheus/client_golang v1.17.0 // indirect 69 69 github.com/prometheus/client_model v0.5.0 // indirect 70 70 github.com/prometheus/common v0.45.0 // indirect 71 71 github.com/prometheus/procfs v0.12.0 // indirect
+9 -2
internal/crawler.go
··· 32 32 // Run enumerates the PDS once at startup, then re-enumerates daily. 33 33 func (c *Crawler) Run(ctx context.Context) { 34 34 for { 35 + start := time.Now() 35 36 if err := c.Enumerate(ctx); err != nil { 36 37 c.logger.Error("failed to enumerate PDS", "error", err) 37 38 } else { 38 39 c.logger.Info("PDS enumeration complete") 39 40 } 41 + crawlerRuns.Inc() 42 + crawlerRunDuration.Observe(time.Since(start).Seconds()) 40 43 41 44 select { 42 45 case <-ctx.Done(): ··· 93 96 repos = append(repos, repo) 94 97 } 95 98 96 - if err := c.db.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(&repos).Error; err != nil { 97 - return fmt.Errorf("failed to save repos batch: %w", err) 99 + result := c.db.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(&repos) 100 + if result.Error != nil { 101 + return fmt.Errorf("failed to save repos batch: %w", result.Error) 102 + } 103 + if n := result.RowsAffected; n > 0 { 104 + crawlerReposDiscovered.Add(float64(n)) 98 105 } 99 106 100 107 total += len(repos)
+35
internal/database.go
··· 1 1 package internal 2 2 3 3 import ( 4 + "context" 5 + 4 6 "gorm.io/driver/sqlite" 5 7 "gorm.io/gorm" 6 8 "gorm.io/gorm/logger" ··· 47 49 type FirehoseCursor struct { 48 50 Url string `gorm:"primaryKey"` 49 51 Seq int64 `gorm:"not null;default:0"` 52 + } 53 + 54 + // transitionRepoStateFromKnown updates a repo's state, emitting a counter 55 + // for the from-> to transition. The caller must pass the previously-loaded 56 + // from state (avoids an extra read where callers already have it). extra is 57 + // merged into the Updates map; pass nil if unused. 58 + func transitionRepoStateFromKnown(ctx context.Context, db *gorm.DB, did string, from RepoState, to RepoState, extra map[string]any) error { 59 + updates := make(map[string]any, len(extra)+1) 60 + for k, v := range extra { 61 + updates[k] = v 62 + } 63 + updates["state"] = to 64 + if err := db.WithContext(ctx).Model(&Repo{}).Where("did = ?", did).Updates(updates).Error; err != nil { 65 + return err 66 + } 67 + if from != to { 68 + reposStateTransitions.WithLabelValues(string(from), string(to)).Inc() 69 + } 70 + return nil 71 + } 72 + 73 + // transitionRepoState reads the current state and transitions the repo, 74 + // emitting a counter for the from-> to transition. Returns nil if the row 75 + // does not exist. 76 + func transitionRepoState(ctx context.Context, db *gorm.DB, did string, to RepoState, extra map[string]any) error { 77 + var r Repo 78 + if err := db.WithContext(ctx).Select("state").First(&r, "did = ?", did).Error; err != nil { 79 + if err == gorm.ErrRecordNotFound { 80 + return nil 81 + } 82 + return err 83 + } 84 + return transitionRepoStateFromKnown(ctx, db, did, r.State, to, extra) 50 85 } 51 86 52 87 func SetupDatabase(dbPath string) (*gorm.DB, error) {
+66 -16
internal/firehose.go
··· 60 60 // resync instead of applying possibly-bad data. 61 61 func (fp *Firehose) ProcessCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error { 62 62 defer fp.updateLastSeq(evt.Seq) 63 + firehoseEventsReceived.WithLabelValues("commit").Inc() 64 + firehoseCommitBytes.Add(float64(len(evt.Blocks))) 63 65 64 66 // serialize with resyncs of this repo; events that arrive mid-resync 65 67 // block here and are re-evaluated against the fresh state afterwards ··· 77 79 fp.logger.Error("failed to auto-track repo", "did", evt.Repo, "error", err) 78 80 return err 79 81 } 82 + firehoseEventsProcessed.WithLabelValues("commit").Inc() 80 83 return nil 81 84 } 82 85 83 86 if curr.State != RepoStateActive { 87 + firehoseEventsSkipped.Inc() 84 88 return nil 85 89 } 86 90 87 91 if curr.Rev != "" && evt.Rev <= curr.Rev { 88 92 fp.logger.Debug("skipping replayed event", "did", evt.Repo, "eventRev", evt.Rev, "currentRev", curr.Rev) 93 + firehoseEventsSkipped.Inc() 89 94 return nil 90 95 } 91 96 92 97 if evt.TooBig { 93 98 fp.logger.Warn("commit event too big, flagging for resync", "did", evt.Repo, "rev", evt.Rev) 99 + firehoseDesync.Inc() 94 100 return fp.markDesynchronized(ctx, evt.Repo) 95 101 } 96 102 97 103 // commits must build directly on our current state 98 104 if evt.PrevData != nil && evt.PrevData.String() != curr.PrevData { 99 105 fp.logger.Warn("repo state desynchronized", "did", evt.Repo, "rev", evt.Rev) 106 + firehoseDesync.Inc() 100 107 return fp.markDesynchronized(ctx, evt.Repo) 101 108 } 102 109 ··· 104 111 repo, err := repolib.VerifyCommitMessage(ctx, evt) 105 112 if err != nil { 106 113 fp.logger.Warn("commit failed structural verification, flagging for resync", "did", evt.Repo, "rev", evt.Rev, "error", err) 114 + firehoseDesync.Inc() 107 115 return fp.markDesynchronized(ctx, evt.Repo) 108 116 } 109 117 110 118 // signature verification against the DID's declared signing key 111 119 if err := repolib.VerifyCommitSignature(ctx, fp.idDir, evt); err != nil { 112 120 fp.logger.Warn("commit failed signature verification, flagging for resync", "did", evt.Repo, "rev", evt.Rev, "error", err) 121 + firehoseDesync.Inc() 113 122 return fp.markDesynchronized(ctx, evt.Repo) 114 123 } 115 124 ··· 122 131 if err != nil { 123 132 // missing CAR file means our disk state is broken; resync 124 133 fp.logger.Error("failed to append commit blocks, flagging for resync", "did", evt.Repo, "error", err) 134 + firehoseDesync.Inc() 125 135 return fp.markDesynchronized(ctx, evt.Repo) 126 136 } 137 + firehoseCommitsAppended.Inc() 138 + firehoseBlocksAppended.Add(float64(nblocks)) 127 139 128 140 if err := fp.db.WithContext(ctx).Model(&Repo{}). 129 141 Where("did = ?", evt.Repo). ··· 136 148 } 137 149 138 150 fp.logger.Debug("appended commit", "did", evt.Repo, "rev", evt.Rev, "blocks", nblocks) 151 + firehoseEventsProcessed.WithLabelValues("commit").Inc() 139 152 return nil 140 153 } 141 154 ··· 143 156 // stream is broken, so flag the repo for a full resync. 144 157 func (fp *Firehose) ProcessSync(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Sync) error { 145 158 defer fp.updateLastSeq(evt.Seq) 159 + firehoseEventsReceived.WithLabelValues("sync").Inc() 146 160 147 161 curr, err := GetRepoState(ctx, fp.db, evt.Did) 148 162 if err != nil { 149 163 return err 150 164 } else if curr == nil { 151 - return EnsureRepo(ctx, fp.db, fp.pdsURL, evt.Did) 165 + if err := EnsureRepo(ctx, fp.db, fp.pdsURL, evt.Did); err != nil { 166 + return err 167 + } 168 + firehoseEventsProcessed.WithLabelValues("sync").Inc() 169 + return nil 152 170 } 153 171 154 172 commit, err := repolib.VerifySyncMessage(ctx, fp.idDir, evt) ··· 157 175 } 158 176 159 177 if curr.State != RepoStateActive { 178 + firehoseEventsSkipped.Inc() 160 179 return nil 161 180 } 162 181 163 182 if curr.Rev != "" && commit.Rev <= curr.Rev { 183 + firehoseEventsSkipped.Inc() 164 184 return nil 165 185 } 166 186 ··· 170 190 } 171 191 172 192 fp.logger.Warn("sync event received, flagging for resync", "did", evt.Did, "rev", commit.Rev) 173 - return fp.markDesynchronized(ctx, evt.Did) 193 + firehoseDesync.Inc() 194 + if err := fp.markDesynchronized(ctx, evt.Did); err != nil { 195 + return err 196 + } 197 + firehoseEventsProcessed.WithLabelValues("sync").Inc() 198 + return nil 174 199 } 175 200 176 201 // ProcessAccount handles #account events: status changes, deletions, and 177 202 // newly created accounts on the PDS. 178 203 func (fp *Firehose) ProcessAccount(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Account) error { 179 204 defer fp.updateLastSeq(evt.Seq) 205 + firehoseEventsReceived.WithLabelValues("account").Inc() 180 206 181 207 curr, err := GetRepoState(ctx, fp.db, evt.Did) 182 208 if err != nil { ··· 184 210 } else if curr == nil { 185 211 if evt.Active { 186 212 fp.logger.Info("discovered new repo from account event", "did", evt.Did) 187 - return EnsureRepo(ctx, fp.db, fp.pdsURL, evt.Did) 213 + if err := EnsureRepo(ctx, fp.db, fp.pdsURL, evt.Did); err != nil { 214 + return err 215 + } 216 + firehoseEventsProcessed.WithLabelValues("account").Inc() 217 + return nil 188 218 } 189 219 return nil 190 220 } ··· 217 247 if err := fp.store.TrashCar(evt.Did); err != nil { 218 248 return fmt.Errorf("failed to trash CAR: %w", err) 219 249 } 220 - return fp.db.WithContext(ctx).Model(&Repo{}). 221 - Where("did = ?", evt.Did). 222 - Updates(map[string]any{ 223 - "state": RepoStateDeleted, 224 - "status": AccountStatusDeleted, 225 - "deleted_at": time.Now().Unix(), 226 - }).Error 250 + if err := transitionRepoStateFromKnown(ctx, fp.db, evt.Did, curr.State, RepoStateDeleted, map[string]any{ 251 + "status": AccountStatusDeleted, 252 + "deleted_at": time.Now().Unix(), 253 + }); err != nil { 254 + return err 255 + } 256 + firehoseEventsProcessed.WithLabelValues("account").Inc() 257 + return nil 227 258 } 228 259 229 260 updates := map[string]any{"status": updateTo} ··· 231 262 // repos parked while the account was unavailable need a fresh fetch 232 263 switch curr.State { 233 264 case RepoStateTakendown, RepoStateSuspended, RepoStateDeactivated, RepoStateDeleted: 234 - updates["state"] = RepoStatePending 235 - updates["deleted_at"] = 0 265 + // state transitions <parked> -> Pending; emit via the helper. 266 + if err := transitionRepoStateFromKnown(ctx, fp.db, evt.Did, curr.State, RepoStatePending, map[string]any{ 267 + "status": updateTo, 268 + "deleted_at": 0, 269 + }); err != nil { 270 + return err 271 + } 272 + firehoseEventsProcessed.WithLabelValues("account").Inc() 273 + return nil 236 274 } 237 275 } 238 - return fp.db.WithContext(ctx).Model(&Repo{}). 276 + if err := fp.db.WithContext(ctx).Model(&Repo{}). 239 277 Where("did = ?", evt.Did). 240 - Updates(updates).Error 278 + Updates(updates).Error; err != nil { 279 + return err 280 + } 281 + firehoseEventsProcessed.WithLabelValues("account").Inc() 282 + return nil 241 283 } 242 284 243 285 // ProcessIdentity handles #identity events. Handles are not part of the 244 286 // repo CAR, so there is nothing to back up; log only. 245 287 func (fp *Firehose) ProcessIdentity(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Identity) error { 246 288 defer fp.updateLastSeq(evt.Seq) 289 + firehoseEventsReceived.WithLabelValues("identity").Inc() 247 290 fp.logger.Debug("identity event", "did", evt.Did) 248 291 return nil 249 292 } ··· 254 297 255 298 func (fp *Firehose) updateLastSeq(seq int64) { 256 299 fp.lastSeq.Store(seq) 300 + firehoseLastSeq.Set(float64(seq)) 257 301 } 258 302 259 303 func (fp *Firehose) saveCursor(ctx context.Context) error { ··· 262 306 return nil 263 307 } 264 308 265 - return fp.db.WithContext(ctx).Save(&FirehoseCursor{ 309 + if err := fp.db.WithContext(ctx).Save(&FirehoseCursor{ 266 310 Url: fp.pdsURL, 267 311 Seq: seq, 268 - }).Error 312 + }).Error; err != nil { 313 + return err 314 + } 315 + firehoseCursorPersistedSeq.Set(float64(seq)) 316 + return nil 269 317 } 270 318 271 319 // RunCursorSaver periodically persists the firehose cursor, and saves one ··· 347 395 return ctx.Err() 348 396 default: 349 397 } 398 + 399 + firehoseReconnects.Inc() 350 400 351 401 cursor, err := fp.GetCursor(ctx) 352 402 if err != nil {
+165
internal/metrics.go
··· 1 + package internal 2 + 3 + import ( 4 + "github.com/prometheus/client_golang/prometheus" 5 + "github.com/prometheus/client_golang/prometheus/promauto" 6 + ) 7 + 8 + // All metrics exposed by the replica service. The metrics server (started by 9 + // cmd/main.go when --metrics-listen is set) serves these at /metrics via 10 + // promhttp. Names are prefixed with replica_ for easy filtering. The set 11 + // mirrors the components in internal/: crawler, resyncer, firehose, car 12 + // store, scrubber, and a poller-driven repos state gauge. 13 + 14 + var ( 15 + // ---- Repo state (gauge sourced by the metrics collector goroutine) ---- 16 + reposTotal = promauto.NewGaugeVec(prometheus.GaugeOpts{ 17 + Name: "replica_repos_total", 18 + Help: "Number of tracked repos, labelled by state and status.", 19 + }, []string{"state", "status"}) 20 + 21 + // Total bytes of CAR files on disk under <data-dir>/repos/. 22 + reposDiskBytes = promauto.NewGauge(prometheus.GaugeOpts{ 23 + Name: "replica_repos_disk_bytes", 24 + Help: "Total size in bytes of all CAR files under <data-dir>/repos/.", 25 + }) 26 + 27 + // ---- State transitions ---- 28 + // Emitted by transitionRepoState (and the bulk resync-reset path) every 29 + // time a repo row's state column changes. 30 + reposStateTransitions = promauto.NewCounterVec(prometheus.CounterOpts{ 31 + Name: "replica_repos_state_transitions_total", 32 + Help: "Total repo state transitions, by from and to state.", 33 + }, []string{"from", "to"}) 34 + 35 + // ---- Crawler ---- 36 + crawlerReposDiscovered = promauto.NewCounter(prometheus.CounterOpts{ 37 + Name: "replica_crawler_repos_discovered_total", 38 + Help: "Total new repos discovered and inserted by the crawler.", 39 + }) 40 + crawlerRuns = promauto.NewCounter(prometheus.CounterOpts{ 41 + Name: "replica_crawler_runs_total", 42 + Help: "Total number of PDS enumeration passes completed.", 43 + }) 44 + crawlerRunDuration = promauto.NewHistogram(prometheus.HistogramOpts{ 45 + Name: "replica_crawler_run_duration_seconds", 46 + Help: "Duration of a single PDS enumeration pass.", 47 + Buckets: prometheus.ExponentialBuckets(1, 2, 12), 48 + }) 49 + 50 + // ---- Resyncer ---- 51 + resyncsStarted = promauto.NewCounter(prometheus.CounterOpts{ 52 + Name: "replica_resyncs_started_total", 53 + Help: "Total repo resync attempts that began a fetch.", 54 + }) 55 + resyncsCompleted = promauto.NewCounter(prometheus.CounterOpts{ 56 + Name: "replica_resyncs_completed_total", 57 + Help: "Total repo resyncs that completed successfully.", 58 + }) 59 + resyncsFailed = promauto.NewCounter(prometheus.CounterOpts{ 60 + Name: "replica_resyncs_failed_total", 61 + Help: "Total repo resyncs that failed.", 62 + }) 63 + resyncsParked = promauto.NewCounterVec(prometheus.CounterOpts{ 64 + Name: "replica_resyncs_parked_total", 65 + Help: "Total repos parked as unavailable (takendown/suspended/deactivated) by the resyncer.", 66 + }, []string{"status"}) 67 + resyncsRateLimited = promauto.NewCounter(prometheus.CounterOpts{ 68 + Name: "replica_resyncs_rate_limited_total", 69 + Help: "Total resync fetches rejected by the PDS with a rate-limit response.", 70 + }) 71 + resyncDuration = promauto.NewHistogram(prometheus.HistogramOpts{ 72 + Name: "replica_resync_duration_seconds", 73 + Help: "Duration of a single repo resync (fetch + verify + write).", 74 + Buckets: prometheus.ExponentialBuckets(0.1, 2, 12), 75 + }) 76 + 77 + // ---- Firehose ---- 78 + firehoseEventsReceived = promauto.NewCounterVec(prometheus.CounterOpts{ 79 + Name: "replica_firehose_events_received_total", 80 + Help: "Total events received from the firehose, by event type.", 81 + }, []string{"type"}) 82 + firehoseEventsProcessed = promauto.NewCounterVec(prometheus.CounterOpts{ 83 + Name: "replica_firehose_events_processed_total", 84 + Help: "Total events fully processed (verified and applied), by event type.", 85 + }, []string{"type"}) 86 + firehoseEventsSkipped = promauto.NewCounter(prometheus.CounterOpts{ 87 + Name: "replica_firehose_events_skipped_total", 88 + Help: "Total events skipped (replayed, or repo not in active state).", 89 + }) 90 + firehoseCommitsAppended = promauto.NewCounter(prometheus.CounterOpts{ 91 + Name: "replica_firehose_commits_appended_total", 92 + Help: "Total commit events whose blocks were appended to a CAR.", 93 + }) 94 + firehoseBlocksAppended = promauto.NewCounter(prometheus.CounterOpts{ 95 + Name: "replica_firehose_blocks_appended_total", 96 + Help: "Total individual blocks appended to CAR files from firehose commits.", 97 + }) 98 + firehoseCommitBytes = promauto.NewCounter(prometheus.CounterOpts{ 99 + Name: "replica_firehose_commit_bytes_total", 100 + Help: "Total bytes of commit slice payloads received from the firehose.", 101 + }) 102 + firehoseDesync = promauto.NewCounter(prometheus.CounterOpts{ 103 + Name: "replica_firehose_desync_total", 104 + Help: "Total times the firehose flagged a repo desynchronized (verification failure or break).", 105 + }) 106 + firehoseLastSeq = promauto.NewGauge(prometheus.GaugeOpts{ 107 + Name: "replica_firehose_last_seq", 108 + Help: "Sequence number of the most recent firehose event seen in memory.", 109 + }) 110 + firehoseCursorPersistedSeq = promauto.NewGauge(prometheus.GaugeOpts{ 111 + Name: "replica_firehose_cursor_persisted_seq", 112 + Help: "Sequence number of the last firehose cursor persisted to the DB.", 113 + }) 114 + firehoseReconnects = promauto.NewCounter(prometheus.CounterOpts{ 115 + Name: "replica_firehose_reconnects_total", 116 + Help: "Total firehose (re)connection attempts, including the initial connect.", 117 + }) 118 + 119 + // ---- CAR store ---- 120 + carAppends = promauto.NewCounter(prometheus.CounterOpts{ 121 + Name: "replica_car_appends_total", 122 + Help: "Total successful AppendCommit operations.", 123 + }) 124 + carAppendBytes = promauto.NewCounter(prometheus.CounterOpts{ 125 + Name: "replica_car_append_bytes_total", 126 + Help: "Total bytes appended to CAR files via AppendCommit.", 127 + }) 128 + carAppendDuration = promauto.NewHistogram(prometheus.HistogramOpts{ 129 + Name: "replica_car_append_duration_seconds", 130 + Help: "Duration of a single AppendCommit operation.", 131 + Buckets: prometheus.ExponentialBuckets(0.001, 2, 12), 132 + }) 133 + carInitDuration = promauto.NewHistogram(prometheus.HistogramOpts{ 134 + Name: "replica_car_init_duration_seconds", 135 + Help: "Duration of a single InitCar (full resync write) operation.", 136 + Buckets: prometheus.ExponentialBuckets(0.01, 2, 12), 137 + }) 138 + carJournalRecoveries = promauto.NewCounter(prometheus.CounterOpts{ 139 + Name: "replica_car_journal_recoveries_total", 140 + Help: "Total root-patch journals replayed by recoverJournal.", 141 + }) 142 + carTrashPurged = promauto.NewCounter(prometheus.CounterOpts{ 143 + Name: "replica_car_trash_purged_total", 144 + Help: "Total trashed CAR backups purged by the scrubber after their retention window.", 145 + }) 146 + 147 + // ---- Scrubber ---- 148 + scrubPasses = promauto.NewCounter(prometheus.CounterOpts{ 149 + Name: "replica_scrub_passes_total", 150 + Help: "Total scrub passes completed by the scrubber.", 151 + }) 152 + scrubPassDuration = promauto.NewHistogram(prometheus.HistogramOpts{ 153 + Name: "replica_scrub_pass_duration_seconds", 154 + Help: "Duration of a single scrub pass over all active repos.", 155 + Buckets: prometheus.ExponentialBuckets(1, 2, 14), 156 + }) 157 + scrubReposChecked = promauto.NewCounter(prometheus.CounterOpts{ 158 + Name: "replica_scrub_repos_checked_total", 159 + Help: "Total repos whose CAR was verified by the scrubber.", 160 + }) 161 + scrubReposFlagged = promauto.NewCounter(prometheus.CounterOpts{ 162 + Name: "replica_scrub_repos_flagged_total", 163 + Help: "Total repos the scrubber flagged desynchronized due to CAR verification failure.", 164 + }) 165 + )
+118
internal/metrics_collector.go
··· 1 + package internal 2 + 3 + import ( 4 + "context" 5 + "log/slog" 6 + "os" 7 + "path/filepath" 8 + "time" 9 + 10 + "gorm.io/gorm" 11 + ) 12 + 13 + // metricsCollector periodically refreshes gauge metrics that are too expensive 14 + // to update on every state change: the per-{state,status} repo count and the 15 + // total size of CAR files on disk. Run as a long-lived goroutine via 16 + // RunMetricsCollector. 17 + type metricsCollector struct { 18 + logger *slog.Logger 19 + db *gorm.DB 20 + dataDir string 21 + interval time.Duration 22 + } 23 + 24 + // RunMetricsCollector polls repo counts and on-disk CAR bytes every 30s, 25 + // updating the replica_repos_total and replica_repos_disk_bytes gauges, until 26 + // ctx is cancelled. 27 + func RunMetricsCollector(ctx context.Context, db *gorm.DB, dataDir string) { 28 + c := &metricsCollector{ 29 + logger: slog.Default().With("component", "metrics_collector"), 30 + db: db, 31 + dataDir: dataDir, 32 + interval: 30 * time.Second, 33 + } 34 + c.run(ctx) 35 + } 36 + 37 + func (c *metricsCollector) run(ctx context.Context) { 38 + // one immediate poll so gauges are populated before the first scrape 39 + c.collect(ctx) 40 + 41 + t := time.NewTicker(c.interval) 42 + defer t.Stop() 43 + for { 44 + select { 45 + case <-ctx.Done(): 46 + return 47 + case <-t.C: 48 + c.collect(ctx) 49 + } 50 + } 51 + } 52 + 53 + // collect refreshes the repos_total gauge vector and repos_disk_bytes gauge. 54 + func (c *metricsCollector) collect(ctx context.Context) { 55 + if err := c.collectRepoCounts(ctx); err != nil { 56 + c.logger.Warn("failed to collect repo counts", "error", err) 57 + } 58 + if err := c.collectDiskBytes(ctx); err != nil { 59 + c.logger.Warn("failed to collect disk bytes", "error", err) 60 + } 61 + } 62 + 63 + func (c *metricsCollector) collectRepoCounts(ctx context.Context) error { 64 + type row struct { 65 + State string 66 + Status string 67 + Count int64 68 + } 69 + var rows []row 70 + if err := c.db.WithContext(ctx). 71 + Raw(`SELECT state AS state, status AS status, COUNT(*) AS count FROM repos GROUP BY state, status`). 72 + Scan(&rows).Error; err != nil { 73 + return err 74 + } 75 + // Reset the vector so label combinations not present in this pass drop to 76 + // zero (rather than lingering with their last value). 77 + reposTotal.Reset() 78 + for _, r := range rows { 79 + reposTotal.WithLabelValues(r.State, r.Status).Set(float64(r.Count)) 80 + } 81 + return nil 82 + } 83 + 84 + // collectDiskBytes walks <dataDir>/repos and sums all *.car file sizes. 85 + func (c *metricsCollector) collectDiskBytes(ctx context.Context) error { 86 + reposDir := filepath.Join(c.dataDir, "repos") 87 + var total int64 88 + err := filepath.WalkDir(reposDir, func(path string, d os.DirEntry, err error) error { 89 + if err != nil { 90 + if os.IsNotExist(err) { 91 + return nil 92 + } 93 + return err 94 + } 95 + if d.IsDir() { 96 + return nil 97 + } 98 + if filepath.Ext(path) != ".car" { 99 + return nil 100 + } 101 + select { 102 + case <-ctx.Done(): 103 + return ctx.Err() 104 + default: 105 + } 106 + fi, err := d.Info() 107 + if err != nil { 108 + return err 109 + } 110 + total += fi.Size() 111 + return nil 112 + }) 113 + if err != nil { 114 + return err 115 + } 116 + reposDiskBytes.Set(float64(total)) 117 + return nil 118 + }
+51 -35
internal/resyncer.go
··· 96 96 } 97 97 98 98 logger.Info("processing resync", "did", did) 99 - if err := r.resyncDid(ctx, did); err != nil { 100 - logger.Error("resync failed", "did", did, "error", err) 99 + resyncsStarted.Inc() 100 + start := time.Now() 101 + resErr := r.resyncDid(ctx, did) 102 + resyncDuration.Observe(time.Since(start).Seconds()) 103 + if resErr != nil { 104 + logger.Error("resync failed", "did", did, "error", resErr) 105 + } else { 106 + resyncsCompleted.Inc() 101 107 } 102 108 } 103 109 } ··· 140 146 if result.RowsAffected == 0 { 141 147 return "", false, nil 142 148 } 149 + // the WHERE clause fixed the from state; emit a transition counter for it. 150 + reposStateTransitions.WithLabelValues(string(state), string(RepoStateResyncing)).Inc() 143 151 return did, true, nil 144 152 } 145 153 ··· 176 184 carBytes, err := comatproto.SyncGetRepo(ctx, client, did, "") 177 185 if err != nil { 178 186 if isRateLimitError(err) { 187 + resyncsRateLimited.Inc() 179 188 r.pdsBackoffMu.Lock() 180 189 r.pdsBackoff[r.pdsURL] = time.Now().Add(10 * time.Second) 181 190 r.pdsBackoffMu.Unlock() ··· 186 195 // instead of retrying. A firehose #account event reactivating it 187 196 // sends it back through the resync queue. 188 197 r.logger.Info("repo unavailable, parking", "did", did, "status", status) 189 - if dbErr := r.db.WithContext(ctx).Model(&Repo{}). 190 - Where("did = ?", did). 191 - Updates(map[string]any{ 192 - "state": RepoState(status), 193 - "status": status, 194 - "error_msg": "", 195 - "retry_count": 0, 196 - "retry_after": 0, 197 - }).Error; dbErr != nil { 198 + // we just claimed it as Resyncing; park via the transition helper so 199 + // the counter records Resyncing -> <parked state>. 200 + if dbErr := transitionRepoStateFromKnown(ctx, r.db, did, RepoStateResyncing, RepoState(status), map[string]any{ 201 + "status": status, 202 + "error_msg": "", 203 + "retry_count": 0, 204 + "retry_after": 0, 205 + }); dbErr != nil { 198 206 return false, fmt.Errorf("failed to park unavailable repo: %w", dbErr) 199 207 } 208 + resyncsParked.WithLabelValues(string(status)).Inc() 200 209 return true, nil 201 210 } 202 211 return false, fmt.Errorf("failed to get repo: %w", err) ··· 222 231 return false, fmt.Errorf("failed to read head CID from CAR: %w", err) 223 232 } 224 233 225 - if err := r.db.WithContext(ctx).Model(&Repo{}). 226 - Where("did = ?", did). 227 - Updates(map[string]any{ 228 - "state": RepoStateActive, 229 - "rev": commit.Rev, 230 - "head": head, 231 - "prev_data": commit.Data.String(), 232 - "error_msg": "", 233 - "retry_count": 0, 234 - "retry_after": 0, 235 - }).Error; err != nil { 234 + // transition Resyncing -> Active (the helper emits the counter). 235 + if err := transitionRepoStateFromKnown(ctx, r.db, did, RepoStateResyncing, RepoStateActive, map[string]any{ 236 + "rev": commit.Rev, 237 + "head": head, 238 + "prev_data": commit.Data.String(), 239 + "error_msg": "", 240 + "retry_count": 0, 241 + "retry_after": 0, 242 + }); err != nil { 236 243 return false, fmt.Errorf("failed to update repo state to active: %w", err) 237 244 } 238 245 ··· 264 271 // start at 1 min, cap at 1 hr between retries 265 272 retryAfter := time.Now().Add(backoff(repo.RetryCount, 60) * 60) 266 273 267 - if err := r.db.WithContext(ctx).Model(&Repo{}). 268 - Where("did = ?", did). 269 - Updates(map[string]any{ 270 - "state": state, 271 - "error_msg": errMsg, 272 - "retry_count": repo.RetryCount + 1, 273 - "retry_after": retryAfter.Unix(), 274 - }).Error; err != nil { 274 + if err := transitionRepoStateFromKnown(ctx, r.db, did, RepoStateResyncing, state, map[string]any{ 275 + "error_msg": errMsg, 276 + "retry_count": repo.RetryCount + 1, 277 + "retry_after": retryAfter.Unix(), 278 + }); err != nil { 275 279 return err 276 280 } 281 + if resyncErr != nil { 282 + resyncsFailed.Inc() 283 + } 277 284 return resyncErr 278 285 } 279 286 280 287 func (r *Resyncer) resetPartiallyResynced(ctx context.Context) error { 281 - return r.db.WithContext(ctx).Model(&Repo{}). 288 + // Bulk-set Resyncing -> Desynchronized for repos orphaned by a crash. We 289 + // can't read-and-update each row, so emit one counter observation per 290 + // affected row using Add rather than reading affected count. 291 + result := r.db.WithContext(ctx).Model(&Repo{}). 282 292 Where("state = ?", RepoStateResyncing). 283 - Update("state", RepoStateDesynchronized).Error 293 + Update("state", RepoStateDesynchronized) 294 + if result.Error != nil { 295 + return result.Error 296 + } 297 + if n := result.RowsAffected; n > 0 { 298 + reposStateTransitions.WithLabelValues(string(RepoStateResyncing), string(RepoStateDesynchronized)).Add(float64(n)) 299 + } 300 + return nil 284 301 } 285 302 286 303 // GetRepoState loads a repo row by DID, returning nil if not tracked. ··· 296 313 } 297 314 298 315 // MarkDesynchronized flags a repo so the resyncer re-fetches its full CAR. 316 + // Emits a state-transition counter; from state is looked up from the DB. 299 317 func MarkDesynchronized(ctx context.Context, db *gorm.DB, did string) error { 300 - return db.WithContext(ctx).Model(&Repo{}). 301 - Where("did = ?", did). 302 - Update("state", RepoStateDesynchronized).Error 318 + return transitionRepoState(ctx, db, did, RepoStateDesynchronized, nil) 303 319 } 304 320 305 321 // EnsureRepo tracks a (usually freshly discovered) repo as pending.
+6
internal/scrubber.go
··· 61 61 func (s *Scrubber) scrubAll(ctx context.Context) error { 62 62 s.logger.Info("starting scrub pass") 63 63 start := time.Now() 64 + defer func() { 65 + scrubPasses.Inc() 66 + scrubPassDuration.Observe(time.Since(start).Seconds()) 67 + }() 64 68 checked := 0 65 69 flagged := 0 66 70 ··· 87 91 if err != nil { 88 92 return err 89 93 } 94 + scrubReposChecked.Inc() 90 95 checked++ 91 96 if !ok { 97 + scrubReposFlagged.Inc() 92 98 flagged++ 93 99 } 94 100 }
+14 -1
internal/store.go
··· 9 9 "os" 10 10 "path/filepath" 11 11 "strings" 12 + "time" 12 13 13 14 "github.com/ipfs/go-cid" 14 15 car "github.com/ipld/go-car" ··· 57 58 // InitCar writes a full repo CAR to disk atomically (temp file + rename), 58 59 // replacing any existing file for the DID. 59 60 func (s *CarStore) InitCar(did string, carBytes []byte) error { 61 + start := time.Now() 62 + defer func() { carInitDuration.Observe(time.Since(start).Seconds()) }() 63 + 60 64 final := s.carPath(did) 61 65 tmp := final + ".tmp" 62 66 ··· 96 100 // declared root is the latest commit. Returns an error if the repo has no 97 101 // CAR file yet. 98 102 func (s *CarStore) AppendCommit(did string, blocks []byte, newHead cid.Cid) (int, error) { 103 + start := time.Now() 99 104 path := s.carPath(did) 100 105 if _, err := os.Stat(path); err != nil { 101 106 return 0, fmt.Errorf("no CAR file for repo %s: %w", did, err) ··· 167 172 return 0, fmt.Errorf("failed to update CAR header root: %w", err) 168 173 } 169 174 175 + carAppends.Inc() 176 + carAppendBytes.Add(float64(len(blocks))) 177 + carAppendDuration.Observe(time.Since(start).Seconds()) 170 178 return n, nil 171 179 } 172 180 ··· 283 291 if err := f.Sync(); err != nil { 284 292 return fmt.Errorf("failed to fsync CAR after journal recovery: %w", err) 285 293 } 286 - return os.Remove(jpath) 294 + if err := os.Remove(jpath); err != nil { 295 + return err 296 + } 297 + carJournalRecoveries.Inc() 298 + return nil 287 299 } 288 300 289 301 // RecoverAllJournals replays any root-patch journals left behind by a crash. ··· 369 381 if err != nil && !os.IsNotExist(err) { 370 382 return fmt.Errorf("failed to delete trashed CAR: %w", err) 371 383 } 384 + carTrashPurged.Inc() 372 385 return nil 373 386 } 374 387