alice.mosphere.at / indigo
this repo has no description
0
fork

Configure Feed

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

don't hold Host struct around so much

bryan newbold b892281d 4a4012a0

+687 -664
+30 -28
cmd/relayered/relay/account.go
··· 15 15 "gorm.io/gorm" 16 16 ) 17 17 18 - // this function with exact name and args implements the `diskpersist.UidSource` interface 19 - func (r *Relay) DidToUid(ctx context.Context, did string) (uint64, error) { 20 - // NOTE: assuming DID is correct syntax (this is usually "loop back") 21 - xu, err := r.GetAccount(ctx, syntax.DID(did)) 22 - if err != nil { 23 - return 0, err 24 - } 25 - if xu == nil { 26 - return 0, ErrAccountNotFound 27 - } 28 - return xu.UID, nil 29 - } 30 - 31 18 func (r *Relay) GetAccount(ctx context.Context, did syntax.DID) (*models.Account, error) { 32 - ctx, span := tracer.Start(ctx, "getAccount") 19 + ctx, span := tracer.Start(ctx, "GetAccount") 33 20 defer span.End() 34 21 35 - /* XXX 36 - cu, ok := r.accountCache.Get(did) 22 + // first try cache 23 + a, ok := r.accountCache.Get(did.String()) 37 24 if ok { 38 - return cu, nil 25 + return a, nil 39 26 } 40 - */ 41 27 42 28 var acc models.Account 43 29 if err := r.db.Where("did = ?", did).First(&acc).Error; err != nil { ··· 47 33 return nil, err 48 34 } 49 35 50 - // TODO: is this further check needed? 36 + // TODO: is this zero UID check redundant? 51 37 if acc.UID == 0 { 52 38 return nil, ErrAccountNotFound 53 39 } 54 40 55 - /* XXX: 56 - r.accountCache.Add(did, &u) 57 - */ 41 + r.accountCache.Add(did.String(), &acc) 58 42 59 43 return &acc, nil 60 44 } ··· 78 62 account, err := r.syncHostAccount(ctx, did, host, nil) 79 63 newUserDiscoveryDuration.Observe(time.Since(start).Seconds()) 80 64 if err != nil { 81 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "uerr").Inc() 82 65 return nil, fmt.Errorf("fed event create external user: %w", err) 83 66 } 84 67 return account, nil ··· 148 131 // we got an event from a non-canonical Host (an intermediate relay) 149 132 // a non-canonical Host we haven't seen before; ping it to make sure it's real 150 133 // TODO: what do we actually want to track about the source we immediately got this message from vs the canonical Host? 151 - r.Logger.Warn("pds discovered in new user flow", "pds", durl.String(), "did", did) 134 + r.Logger.Warn("new host discovered in create user flow", "pds", durl.String(), "did", did) 152 135 153 136 err = r.HostChecker.CheckHost(ctx, durl.String()) 154 137 if err != nil { ··· 183 166 } 184 167 185 168 // this lock just governs the lower half of this function 186 - r.extUserLk.Lock() 187 - defer r.extUserLk.Unlock() 169 + r.accountLk.Lock() 170 + defer r.accountLk.Unlock() 188 171 189 172 if cachedAccount == nil { 190 173 cachedAccount, err = r.GetAccount(ctx, did) ··· 206 189 tx.Model(&models.Host{}).Where("id = ?", caHost).Update("account_count", gorm.Expr("account_count - 1")) 207 190 } 208 191 // update user's Host ID 209 - res := tx.Model(models.Account{}).Where("id = ?", cachedAccount.UID).Update("pds", canonicalHost.ID) 192 + res := tx.Model(models.Account{}).Where("uid = ?", cachedAccount.UID).Update("host_id", canonicalHost.ID) 210 193 if res.Error != nil { 211 194 return fmt.Errorf("failed to update users pds: %w", res.Error) 212 195 } ··· 217 200 218 201 // XXX: cachedAccount.SetHost(canonicalHost.ID) 219 202 cachedAccount.HostID = canonicalHost.ID 203 + 204 + // flush account cache 205 + r.accountCache.Remove(did.String()) 220 206 } 221 207 return cachedAccount, nil 222 208 } ··· 259 245 return err 260 246 } 261 247 248 + // clear account cache 249 + r.accountCache.Remove(did.String()) 250 + 262 251 // NOTE: not wiping events for user from persister (backfill window) 263 252 return nil 264 253 } ··· 266 255 func (r *Relay) ListAccounts(ctx context.Context, cursor int64, limit int) ([]*models.Account, error) { 267 256 268 257 accounts := []*models.Account{} 269 - if err := r.db.Model(&models.Account{}).Where("id > ? AND NOT taken_down AND (upstream_status IS NULL OR upstream_status = 'active')", cursor).Order("id").Limit(limit).Find(&accounts).Error; err != nil { 258 + if err := r.db.Model(&models.Account{}).Where("uid > ? AND status IS NOT 'takendown' AND (upstream_status IS NULL OR upstream_status = 'active')", cursor).Order("uid").Limit(limit).Find(&accounts).Error; err != nil { 270 259 return nil, err 271 260 } 272 261 return accounts, nil ··· 275 264 func (r *Relay) UpsertAccountRepo(uid uint64, rev syntax.TID, commitCID, commitDataCID cid.Cid) error { 276 265 return r.db.Exec("INSERT INTO account_repo (uid, rev, commit_cid, commit_data) VALUES (?, ?, ?, ?) ON CONFLICT (uid) DO UPDATE SET rev = EXCLUDED.rev, commit_cid = EXCLUDED.commit_cid, commit_data = EXCLUDED.commit_data", uid, rev, commitCID.String(), commitDataCID.String()).Error 277 266 } 267 + 268 + // this function with exact name and args implements the `diskpersist.UidSource` interface 269 + func (r *Relay) DidToUid(ctx context.Context, did string) (uint64, error) { 270 + // NOTE: not re-parsing DID here (this function is called "loopback" from persister) 271 + xu, err := r.GetAccount(ctx, syntax.DID(did)) 272 + if err != nil { 273 + return 0, err 274 + } 275 + if xu == nil { 276 + return 0, ErrAccountNotFound 277 + } 278 + return xu.UID, nil 279 + }
+162 -302
cmd/relayered/relay/firehose.go
··· 2 2 3 3 import ( 4 4 "context" 5 - "errors" 6 5 "fmt" 6 + "net" 7 + "net/http" 8 + "sync" 7 9 "time" 8 10 9 - comatproto "github.com/bluesky-social/indigo/api/atproto" 10 - "github.com/bluesky-social/indigo/atproto/syntax" 11 - "github.com/bluesky-social/indigo/cmd/relayered/relay/models" 12 11 "github.com/bluesky-social/indigo/cmd/relayered/stream" 13 12 14 - "github.com/ipfs/go-cid" 15 - "go.opentelemetry.io/otel/attribute" 16 - "gorm.io/gorm" 13 + "github.com/gorilla/websocket" 14 + promclient "github.com/prometheus/client_golang/prometheus" 15 + dto "github.com/prometheus/client_model/go" 17 16 ) 18 17 19 - // handleFedEvent() is the callback passed to Slurper called from Slurper.handleConnection() 20 - // XXX: evt not env 21 - func (r *Relay) handleFedEvent(ctx context.Context, host *models.Host, env *stream.XRPCStreamEvent) error { 22 - ctx, span := tracer.Start(ctx, "handleFedEvent") 23 - defer span.End() 18 + type SocketConsumer struct { 19 + UserAgent string 20 + RemoteAddr string 21 + ConnectedAt time.Time 22 + EventsSent promclient.Counter 23 + } 24 24 25 - start := time.Now() 26 - defer func() { 27 - eventsHandleDuration.WithLabelValues(host.Hostname).Observe(time.Since(start).Seconds()) 28 - }() 25 + func (r *Relay) registerConsumer(c *SocketConsumer) uint64 { 26 + r.consumersLk.Lock() 27 + defer r.consumersLk.Unlock() 29 28 30 - EventsReceivedCounter.WithLabelValues(host.Hostname).Add(1) 29 + id := r.nextConsumerID 30 + r.nextConsumerID++ 31 31 32 - switch { 33 - case env.RepoCommit != nil: 34 - repoCommitsReceivedCounter.WithLabelValues(host.Hostname).Add(1) 35 - return r.handleCommit(ctx, host, env.RepoCommit) 36 - case env.RepoSync != nil: 37 - repoSyncReceivedCounter.WithLabelValues(host.Hostname).Add(1) 38 - return r.handleSync(ctx, host, env.RepoSync) 39 - case env.RepoHandle != nil: 40 - eventsWarningsCounter.WithLabelValues(host.Hostname, "handle").Add(1) 41 - // TODO: rate limit warnings per Host before we (temporarily?) block them 42 - return nil 43 - case env.RepoIdentity != nil: 44 - r.Logger.Info("relay got identity event", "did", env.RepoIdentity.Did) 32 + r.consumers[id] = c 45 33 46 - did, err := syntax.ParseDID(env.RepoIdentity.Did) 47 - if err != nil { 48 - return fmt.Errorf("invalid DID in message: %w", err) 49 - } 34 + return id 35 + } 50 36 51 - // Flush any cached DID documents for this user 52 - r.purgeDidCache(ctx, did.String()) 37 + func (r *Relay) cleanupConsumer(id uint64) { 38 + r.consumersLk.Lock() 39 + defer r.consumersLk.Unlock() 53 40 54 - // Refetch the DID doc and update our cached keys and handle etc. 55 - account, err := r.syncHostAccount(ctx, did, host, nil) 56 - if err != nil { 57 - return err 58 - } 41 + c := r.consumers[id] 59 42 60 - // Broadcast the identity event to all consumers 61 - err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 62 - RepoIdentity: &comatproto.SyncSubscribeRepos_Identity{ 63 - Did: did.String(), 64 - Seq: env.RepoIdentity.Seq, 65 - Time: env.RepoIdentity.Time, 66 - Handle: env.RepoIdentity.Handle, 67 - }, 68 - PrivUid: account.UID, 69 - }) 70 - if err != nil { 71 - r.Logger.Error("failed to broadcast Identity event", "error", err, "did", did) 72 - return fmt.Errorf("failed to broadcast Identity event: %w", err) 73 - } 43 + var m = &dto.Metric{} 44 + if err := c.EventsSent.Write(m); err != nil { 45 + r.Logger.Error("failed to get sent counter", "err", err) 46 + } 74 47 75 - return nil 76 - case env.RepoAccount != nil: 77 - span.SetAttributes( 78 - attribute.String("did", env.RepoAccount.Did), 79 - attribute.Int64("seq", env.RepoAccount.Seq), 80 - attribute.Bool("active", env.RepoAccount.Active), 81 - ) 48 + r.Logger.Info("consumer disconnected", 49 + "consumer_id", id, 50 + "remote_addr", c.RemoteAddr, 51 + "user_agent", c.UserAgent, 52 + "events_sent", m.Counter.GetValue()) 82 53 83 - did, err := syntax.ParseDID(env.RepoAccount.Did) 84 - if err != nil { 85 - return fmt.Errorf("invalid DID in message: %w", err) 86 - } 54 + delete(r.consumers, id) 55 + } 87 56 88 - if env.RepoAccount.Status != nil { 89 - span.SetAttributes(attribute.String("repo_status", *env.RepoAccount.Status)) 90 - } 91 - r.Logger.Info("relay got account event", "did", env.RepoAccount.Did) 57 + // Main HTTP request handler for clients connecting to the firehose (com.atproto.sync.subscribeRepos) 58 + func (r *Relay) HandleSubscribeRepos(resp http.ResponseWriter, req *http.Request, since *int64, realIP string) error { 92 59 93 - if !env.RepoAccount.Active && env.RepoAccount.Status == nil { 94 - // TODO: semantics here aren't really clear 95 - r.Logger.Warn("dropping invalid account event", "did", env.RepoAccount.Did, "active", env.RepoAccount.Active, "status", env.RepoAccount.Status) 96 - accountVerifyWarnings.WithLabelValues(host.Hostname, "nostat").Inc() 97 - return nil 98 - } 60 + ctx, cancel := context.WithCancel(req.Context()) 61 + defer cancel() 99 62 100 - // Flush any cached DID documents for this user 101 - r.purgeDidCache(ctx, did.String()) 63 + conn, err := websocket.Upgrade(resp, req, resp.Header(), 10<<10, 10<<10) 64 + if err != nil { 65 + return fmt.Errorf("upgrading websocket: %w", err) 66 + } 102 67 103 - // Refetch the DID doc to make sure the Host is still authoritative 104 - account, err := r.syncHostAccount(ctx, did, host, nil) 105 - if err != nil { 106 - span.RecordError(err) 107 - return err 108 - } 68 + defer conn.Close() 109 69 110 - // Check if the Host is still authoritative 111 - // if not we don't want to be propagating this account event 112 - // XXX: lock 113 - if account.HostID != host.ID && !r.Config.SkipAccountHostCheck { 114 - r.Logger.Error("account event from non-authoritative pds", 115 - "seq", env.RepoAccount.Seq, 116 - "did", env.RepoAccount.Did, 117 - "event_from", host.Hostname, 118 - "did_doc_declared_pds", account.HostID, 119 - "account_evt", env.RepoAccount, 120 - ) 121 - return fmt.Errorf("event from non-authoritative pds") 122 - } 70 + lastWriteLk := sync.Mutex{} 71 + lastWrite := time.Now() 123 72 124 - // Process the account status change 125 - repoStatus := models.AccountStatusActive 126 - if !env.RepoAccount.Active && env.RepoAccount.Status != nil { 127 - repoStatus = models.AccountStatus(*env.RepoAccount.Status) 128 - } 73 + // Start a goroutine to ping the client every 30 seconds to check if it's 74 + // still alive. If the client doesn't respond to a ping within 5 seconds, 75 + // we'll close the connection and teardown the consumer. 76 + go func() { 77 + ticker := time.NewTicker(30 * time.Second) 78 + defer ticker.Stop() 129 79 130 - // XXX: lock, and parse 131 - account.UpstreamStatus = models.AccountStatus(repoStatus) 132 - err = r.db.Save(account).Error 133 - if err != nil { 134 - span.RecordError(err) 135 - return fmt.Errorf("failed to update account status: %w", err) 136 - } 80 + for { 81 + select { 82 + case <-ticker.C: 83 + lastWriteLk.Lock() 84 + lw := lastWrite 85 + lastWriteLk.Unlock() 137 86 138 - shouldBeActive := env.RepoAccount.Active 139 - status := env.RepoAccount.Status 87 + if time.Since(lw) < 30*time.Second { 88 + continue 89 + } 140 90 141 - // override with local status 142 - // XXX: lock 143 - if account.Status == "takendown" { 144 - shouldBeActive = false 145 - s := string(models.AccountStatusTakendown) 146 - status = &s 91 + if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil { 92 + r.Logger.Warn("failed to ping client", "err", err) 93 + cancel() 94 + return 95 + } 96 + case <-ctx.Done(): 97 + return 98 + } 147 99 } 100 + }() 148 101 149 - // Broadcast the account event to all consumers 150 - err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 151 - RepoAccount: &comatproto.SyncSubscribeRepos_Account{ 152 - Active: shouldBeActive, 153 - Did: env.RepoAccount.Did, 154 - Seq: env.RepoAccount.Seq, 155 - Status: status, 156 - Time: env.RepoAccount.Time, 157 - }, 158 - PrivUid: account.UID, 159 - }) 160 - if err != nil { 161 - r.Logger.Error("failed to broadcast Account event", "error", err, "did", env.RepoAccount.Did) 162 - return fmt.Errorf("failed to broadcast Account event: %w", err) 102 + conn.SetPingHandler(func(message string) error { 103 + err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60)) 104 + if err == websocket.ErrCloseSent { 105 + return nil 106 + } else if e, ok := err.(net.Error); ok && e.Temporary() { 107 + return nil 163 108 } 109 + return err 110 + }) 164 111 165 - return nil 166 - case env.RepoMigrate != nil: 167 - eventsWarningsCounter.WithLabelValues(host.Hostname, "migrate").Add(1) 168 - // TODO: rate limit warnings per Host before we (temporarily?) block them 169 - return nil 170 - case env.RepoTombstone != nil: 171 - eventsWarningsCounter.WithLabelValues(host.Hostname, "tombstone").Add(1) 172 - // TODO: rate limit warnings per Host before we (temporarily?) block them 173 - return nil 174 - default: 175 - return fmt.Errorf("invalid fed event") 176 - } 177 - } 112 + // Start a goroutine to read messages from the client and discard them. 113 + go func() { 114 + for { 115 + _, _, err := conn.ReadMessage() 116 + if err != nil { 117 + r.Logger.Warn("failed to read message from client", "err", err) 118 + cancel() 119 + return 120 + } 121 + } 122 + }() 178 123 179 - func (r *Relay) handleCommit(ctx context.Context, host *models.Host, evt *comatproto.SyncSubscribeRepos_Commit) error { 180 - r.Logger.Debug("relay got repo append event", "seq", evt.Seq, "host", host.Hostname, "repo", evt.Repo) 124 + ident := realIP + "-" + req.UserAgent() 181 125 182 - did, err := syntax.ParseDID(evt.Repo) 126 + evts, cleanup, err := r.Events.Subscribe(ctx, ident, func(evt *stream.XRPCStreamEvent) bool { return true }, since) 183 127 if err != nil { 184 - return fmt.Errorf("invalid DID in message: %w", err) 128 + return err 185 129 } 186 - // XXX: did = did.Normalize() 187 - account, err := r.GetAccount(ctx, did) 188 - if err != nil { 189 - if !errors.Is(err, gorm.ErrRecordNotFound) { 190 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "nou").Inc() 191 - return fmt.Errorf("looking up event user: %w", err) 192 - } 130 + defer cleanup() 193 131 194 - account, err = r.CreateAccount(ctx, host, did) 195 - if err != nil { 196 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "nuerr").Inc() 197 - return err 198 - } 199 - } 200 - if account == nil { 201 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "nou2").Inc() 202 - return ErrAccountNotFound 132 + // Keep track of the consumer for metrics and admin endpoints 133 + consumer := SocketConsumer{ 134 + RemoteAddr: realIP, 135 + UserAgent: req.UserAgent(), 136 + ConnectedAt: time.Now(), 203 137 } 138 + sentCounter := eventsSentCounter.WithLabelValues(consumer.RemoteAddr, consumer.UserAgent) 139 + consumer.EventsSent = sentCounter 204 140 205 - // XXX: lock on account 206 - ustatus := account.UpstreamStatus 141 + consumerID := r.registerConsumer(&consumer) 142 + defer r.cleanupConsumer(consumerID) 207 143 208 - // XXX: lock on account 209 - if account.Status == models.AccountStatusTakendown || ustatus == models.AccountStatusTakendown { 210 - r.Logger.Debug("dropping commit event from taken down user", "did", evt.Repo, "seq", evt.Seq, "host", host.Hostname) 211 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "tdu").Inc() 212 - return nil 213 - } 144 + logger := r.Logger.With( 145 + "consumer_id", consumerID, 146 + "remote_addr", consumer.RemoteAddr, 147 + "user_agent", consumer.UserAgent, 148 + ) 214 149 215 - if ustatus == models.AccountStatusSuspended { 216 - r.Logger.Debug("dropping commit event from suspended user", "did", evt.Repo, "seq", evt.Seq, "host", host.Hostname) 217 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "susu").Inc() 218 - return nil 219 - } 220 - 221 - if ustatus == models.AccountStatusDeactivated { 222 - r.Logger.Debug("dropping commit event from deactivated user", "did", evt.Repo, "seq", evt.Seq, "host", host.Hostname) 223 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "du").Inc() 224 - return nil 225 - } 150 + logger.Info("new consumer", "cursor", since) 226 151 227 - if evt.Rebase { 228 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "rebase").Inc() 229 - return fmt.Errorf("rebase was true in event seq:%d,host:%s", evt.Seq, host.Hostname) 230 - } 152 + for { 153 + select { 154 + case evt, ok := <-evts: 155 + if !ok { 156 + logger.Error("event stream closed unexpectedly") 157 + return nil 158 + } 231 159 232 - accountHostId := account.HostID 233 - if host.ID != accountHostId && accountHostId != 0 { 234 - r.Logger.Warn("received event for repo from different pds than expected", "repo", evt.Repo, "expPds", accountHostId, "gotPds", host.Hostname) 235 - // Flush any cached DID documents for this user 236 - r.purgeDidCache(ctx, evt.Repo) 160 + wc, err := conn.NextWriter(websocket.BinaryMessage) 161 + if err != nil { 162 + logger.Error("failed to get next writer", "err", err) 163 + return err 164 + } 237 165 238 - account, err = r.syncHostAccount(ctx, did, host, account) 239 - if err != nil { 240 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "uerr2").Inc() 241 - return err 242 - } 166 + if evt.Preserialized != nil { 167 + _, err = wc.Write(evt.Preserialized) 168 + } else { 169 + err = evt.Serialize(wc) 170 + } 171 + if err != nil { 172 + return fmt.Errorf("failed to write event: %w", err) 173 + } 243 174 244 - if account.HostID != host.ID && !r.Config.SkipAccountHostCheck { 245 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "noauth").Inc() 246 - return fmt.Errorf("event from non-authoritative pds") 247 - } 248 - } 175 + if err := wc.Close(); err != nil { 176 + logger.Warn("failed to flush-close our event write", "err", err) 177 + return nil 178 + } 249 179 250 - // TODO: very messy fetch code here 251 - var repo *models.AccountRepo 252 - err = r.db.First(repo, account.UID).Error 253 - if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { 254 - r.Logger.Error("failed to get previous root", "err", err) 255 - repo = nil 256 - } 257 - var prevRev *syntax.TID 258 - var prevData *cid.Cid 259 - if repo != nil { 260 - c, err := cid.Parse(repo.CommitData) 261 - if err != nil { 262 - return fmt.Errorf("parsing commitDataCID from database: %w", err) 180 + lastWriteLk.Lock() 181 + lastWrite = time.Now() 182 + lastWriteLk.Unlock() 183 + sentCounter.Inc() 184 + case <-ctx.Done(): 185 + return nil 263 186 } 264 - prevData = &c 265 - t := syntax.TID(repo.Rev) 266 - prevRev = &t 267 187 } 268 - evtPrevDataStr := "" 269 - if evt.PrevData != nil { 270 - evtPrevDataStr = ((*cid.Cid)(evt.PrevData)).String() 271 - } 272 - commitDataCID, err := r.Validator.HandleCommit(ctx, host, account, evt, prevRev, prevData) 273 - if err != nil { 274 - // XXX: induction trace log 275 - r.Logger.Error("commit bad", "seq", evt.Seq, "host", host.Hostname, "repo", evt.Repo, "prev", evtPrevDataStr, "err", err) 276 - r.Logger.Warn("failed handling event", "err", err, "host", host.Hostname, "seq", evt.Seq, "repo", account.DID, "commit", evt.Commit.String()) 277 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "err").Inc() 278 - return fmt.Errorf("handle user event failed: %w", err) 279 - } 280 - 281 - // TID syntax has been verified by validator 282 - rev := syntax.TID(evt.Rev) 283 - 284 - err = r.UpsertAccountRepo(account.UID, rev, cid.Cid(evt.Commit), *commitDataCID) 285 - if err != nil { 286 - return fmt.Errorf("failed to set previous root uid=%d: %w", account.UID, err) 287 - } 288 - 289 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "ok").Inc() 290 - 291 - // Broadcast the identity event to all consumers 292 - commitCopy := *evt 293 - err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 294 - RepoCommit: &commitCopy, 295 - PrivUid: account.UID, 296 - }) 297 - if err != nil { 298 - r.Logger.Error("failed to broadcast commit event", "error", err, "did", evt.Repo) 299 - return fmt.Errorf("failed to broadcast commit event: %w", err) 300 - } 301 - 302 - return nil 303 188 } 304 189 305 - // handleSync processes #sync messages 306 - func (r *Relay) handleSync(ctx context.Context, host *models.Host, evt *comatproto.SyncSubscribeRepos_Sync) error { 307 - did, err := syntax.ParseDID(evt.Did) 308 - if err != nil { 309 - return fmt.Errorf("invalid DID in message: %s", did) 310 - } 311 - // XXX: did.Normalize() 312 - account, err := r.GetAccount(ctx, did) 313 - if err != nil { 314 - if !errors.Is(err, gorm.ErrRecordNotFound) { 315 - repoCommitsResultCounter.WithLabelValues(host.Hostname, "nou").Inc() 316 - return fmt.Errorf("looking up event user: %w", err) 317 - } 318 - 319 - account, err = r.CreateAccount(ctx, host, did) 320 - } 321 - if err != nil { 322 - return fmt.Errorf("could not get user for did %#v: %w", evt.Did, err) 323 - } 324 - 325 - commitCID, commitDataCID, err := r.Validator.HandleSync(ctx, host, evt) 326 - if err != nil { 327 - return err 328 - } 329 - // TID syntax has been verified by validator 330 - rev := syntax.TID(evt.Rev) 331 - 332 - // TODO: should this happen before or after firehose persist/broadcast? 333 - err = r.UpsertAccountRepo(account.UID, rev, *commitCID, *commitDataCID) 334 - if err != nil { 335 - return fmt.Errorf("failed to upsert repo state (uid %d): %w", account.UID, err) 336 - } 337 - 338 - // Broadcast the sync event to all consumers 339 - evtCopy := *evt 340 - err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 341 - RepoSync: &evtCopy, 342 - }) 343 - if err != nil { 344 - r.Logger.Error("failed to broadcast sync event", "error", err, "did", evt.Did) 345 - return fmt.Errorf("failed to broadcast sync event: %w", err) 346 - } 347 - 348 - return nil 190 + type ConsumerInfo struct { 191 + ID uint64 `json:"id"` 192 + RemoteAddr string `json:"remote_addr"` 193 + UserAgent string `json:"user_agent"` 194 + EventsConsumed uint64 `json:"events_consumed"` 195 + ConnectedAt time.Time `json:"connected_at"` 349 196 } 350 197 351 - func (r *Relay) purgeDidCache(ctx context.Context, did string) { 352 - ati, err := syntax.ParseAtIdentifier(did) 353 - if err != nil { 354 - return 198 + func (r *Relay) ListConsumers() []ConsumerInfo { 199 + r.consumersLk.RLock() 200 + defer r.consumersLk.RUnlock() 201 + 202 + info := make([]ConsumerInfo, 0, len(r.consumers)) 203 + for id, c := range r.consumers { 204 + var m = &dto.Metric{} 205 + if err := c.EventsSent.Write(m); err != nil { 206 + continue 207 + } 208 + info = append(info, ConsumerInfo{ 209 + ID: id, 210 + RemoteAddr: c.RemoteAddr, 211 + UserAgent: c.UserAgent, 212 + EventsConsumed: uint64(m.Counter.GetValue()), 213 + ConnectedAt: c.ConnectedAt, 214 + }) 355 215 } 356 - _ = r.dir.Purge(ctx, *ati) 216 + return info 357 217 }
+384
cmd/relayered/relay/ingest.go
··· 1 + package relay 2 + 3 + import ( 4 + "context" 5 + "errors" 6 + "fmt" 7 + "time" 8 + 9 + comatproto "github.com/bluesky-social/indigo/api/atproto" 10 + "github.com/bluesky-social/indigo/atproto/syntax" 11 + "github.com/bluesky-social/indigo/cmd/relayered/relay/models" 12 + "github.com/bluesky-social/indigo/cmd/relayered/stream" 13 + 14 + "github.com/ipfs/go-cid" 15 + "go.opentelemetry.io/otel/attribute" 16 + "gorm.io/gorm" 17 + ) 18 + 19 + // This callback function gets called by Slurper on every upstream repo stream message from any host. 20 + // 21 + // Messages are processed in-order for a single account on a single host; but may be concurrent or out-of-order for the same account *across* hosts (eg, during account migration or a conflict) 22 + func (r *Relay) processRepoEvent(ctx context.Context, evt *stream.XRPCStreamEvent, hostname string, hostID uint64) error { 23 + ctx, span := tracer.Start(ctx, "processRepoEvent") 24 + defer span.End() 25 + 26 + start := time.Now() 27 + defer func() { 28 + eventsHandleDuration.WithLabelValues(hostname).Observe(time.Since(start).Seconds()) 29 + }() 30 + 31 + EventsReceivedCounter.WithLabelValues(hostname).Add(1) 32 + 33 + switch { 34 + case evt.RepoCommit != nil: 35 + repoCommitsReceivedCounter.WithLabelValues(hostname).Add(1) 36 + return r.processCommitEvent(ctx, evt.RepoCommit, hostname, hostID) 37 + case evt.RepoSync != nil: 38 + repoSyncReceivedCounter.WithLabelValues(hostname).Add(1) 39 + return r.processSyncEvent(ctx, evt.RepoSync, hostname, hostID) 40 + case evt.RepoIdentity != nil: 41 + //repoIdentityReceivedCounter.WithLabelValues(hostname).Add(1) 42 + return r.processIdentityEvent(ctx, evt.RepoIdentity, hostname, hostID) 43 + case evt.RepoAccount != nil: 44 + //repoAccountReceivedCounter.WithLabelValues(hostname).Add(1) 45 + return r.processAccountEvent(ctx, evt.RepoAccount, hostname, hostID) 46 + case evt.RepoHandle != nil: // DEPRECATED 47 + eventsWarningsCounter.WithLabelValues(hostname, "handle").Add(1) 48 + return nil 49 + case evt.RepoMigrate != nil: // DEPRECATED 50 + eventsWarningsCounter.WithLabelValues(hostname, "migrate").Add(1) 51 + return nil 52 + case evt.RepoTombstone != nil: // DEPRECATED 53 + eventsWarningsCounter.WithLabelValues(hostname, "tombstone").Add(1) 54 + return nil 55 + default: 56 + return fmt.Errorf("unhandled repo stream event type") 57 + } 58 + } 59 + 60 + func (r *Relay) processCommitEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit, hostname string, hostID uint64) error { 61 + logger := r.Logger.With("did", evt.Repo, "seq", evt.Seq, "host", hostname, "eventType", "commit", "rev", evt.Rev) 62 + logger.Debug("relay got repo append event") 63 + 64 + did, err := syntax.ParseDID(evt.Repo) 65 + if err != nil { 66 + return fmt.Errorf("invalid DID in message: %w", err) 67 + } 68 + // XXX: did = did.Normalize() 69 + account, err := r.GetAccount(ctx, did) 70 + if err != nil { 71 + if !errors.Is(err, gorm.ErrRecordNotFound) { 72 + return fmt.Errorf("looking up event user: %w", err) 73 + } 74 + 75 + host, err := r.GetHost(ctx, hostID) 76 + if err != nil { 77 + return err 78 + } 79 + account, err = r.CreateAccount(ctx, host, did) 80 + if err != nil { 81 + return err 82 + } 83 + } 84 + if account == nil { 85 + return ErrAccountNotFound 86 + } 87 + 88 + // XXX: lock on account 89 + ustatus := account.UpstreamStatus 90 + 91 + // XXX: lock on account 92 + if account.Status == models.AccountStatusTakendown || ustatus == models.AccountStatusTakendown { 93 + logger.Debug("dropping commit event from taken down user") 94 + return nil 95 + } 96 + 97 + if ustatus == models.AccountStatusSuspended { 98 + logger.Debug("dropping commit event from suspended user") 99 + return nil 100 + } 101 + 102 + if ustatus == models.AccountStatusDeactivated { 103 + logger.Debug("dropping commit event from deactivated user") 104 + return nil 105 + } 106 + 107 + if evt.Rebase { 108 + return fmt.Errorf("rebase was true in event seq:%d,host:%s", evt.Seq, hostname) 109 + } 110 + 111 + accountHostId := account.HostID 112 + if hostID != accountHostId && accountHostId != 0 { 113 + // XXX: metter logging 114 + logger.Warn("received event for repo from different pds than expected", "expectedHostID", accountHostId, "receivedHost", hostname) 115 + // Flush any cached DID documents for this user 116 + err = r.dir.Purge(ctx, did.AtIdentifier()) 117 + if err != nil { 118 + logger.Error("problem purging identity directory cache", "err", err) 119 + } 120 + 121 + // XXX: shouldn't need full Host? 122 + host, err := r.GetHost(ctx, hostID) 123 + if err != nil { 124 + return err 125 + } 126 + 127 + account, err = r.syncHostAccount(ctx, did, host, account) 128 + if err != nil { 129 + return err 130 + } 131 + 132 + if account.HostID != hostID && !r.Config.SkipAccountHostCheck { 133 + return fmt.Errorf("event from non-authoritative pds") 134 + } 135 + } 136 + 137 + // TODO: very messy fetch code here 138 + var repo *models.AccountRepo 139 + err = r.db.First(repo, account.UID).Error 140 + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { 141 + logger.Error("failed to get previous root", "err", err) 142 + repo = nil 143 + } 144 + var prevRev *syntax.TID 145 + var prevData *cid.Cid 146 + if repo != nil { 147 + c, err := cid.Parse(repo.CommitData) 148 + if err != nil { 149 + return fmt.Errorf("parsing commitDataCID from database: %w", err) 150 + } 151 + prevData = &c 152 + t := syntax.TID(repo.Rev) 153 + prevRev = &t 154 + } 155 + evtPrevDataStr := "" 156 + if evt.PrevData != nil { 157 + evtPrevDataStr = ((*cid.Cid)(evt.PrevData)).String() 158 + } 159 + commitDataCID, err := r.Validator.HandleCommit(ctx, hostname, account, evt, prevRev, prevData) 160 + if err != nil { 161 + // XXX: induction trace log 162 + logger.Error("commit bad", "prevData", evtPrevDataStr, "err", err) 163 + logger.Warn("failed handling event", "err", err, "commitCID", evt.Commit.String()) 164 + return fmt.Errorf("handle user event failed: %w", err) 165 + } 166 + 167 + // TID syntax has been verified by validator 168 + rev := syntax.TID(evt.Rev) 169 + 170 + err = r.UpsertAccountRepo(account.UID, rev, cid.Cid(evt.Commit), *commitDataCID) 171 + if err != nil { 172 + return fmt.Errorf("failed to set previous root uid=%d: %w", account.UID, err) 173 + } 174 + 175 + // Broadcast the identity event to all consumers 176 + commitCopy := *evt 177 + err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 178 + RepoCommit: &commitCopy, 179 + PrivUid: account.UID, 180 + }) 181 + if err != nil { 182 + logger.Error("failed to broadcast commit event", "error", err) 183 + return fmt.Errorf("failed to broadcast commit event: %w", err) 184 + } 185 + 186 + return nil 187 + } 188 + 189 + func (r *Relay) processSyncEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Sync, hostname string, hostID uint64) error { 190 + logger := r.Logger.With("did", evt.Did, "seq", evt.Seq, "host", hostname, "eventType", "sync") 191 + did, err := syntax.ParseDID(evt.Did) 192 + if err != nil { 193 + return fmt.Errorf("invalid DID in message: %s", did) 194 + } 195 + // XXX: did.Normalize() 196 + account, err := r.GetAccount(ctx, did) 197 + if err != nil { 198 + if !errors.Is(err, gorm.ErrRecordNotFound) { 199 + return fmt.Errorf("looking up event user: %w", err) 200 + } 201 + 202 + host, err := r.GetHost(ctx, hostID) 203 + if err != nil { 204 + return err 205 + } 206 + account, err = r.CreateAccount(ctx, host, did) 207 + } 208 + if err != nil { 209 + return fmt.Errorf("could not get user for did %#v: %w", evt.Did, err) 210 + } 211 + 212 + commitCID, commitDataCID, err := r.Validator.HandleSync(ctx, hostname, evt) 213 + if err != nil { 214 + return err 215 + } 216 + // TID syntax has been verified by validator 217 + rev := syntax.TID(evt.Rev) 218 + 219 + // TODO: should this happen before or after firehose persist/broadcast? 220 + err = r.UpsertAccountRepo(account.UID, rev, *commitCID, *commitDataCID) 221 + if err != nil { 222 + return fmt.Errorf("failed to upsert repo state (uid %d): %w", account.UID, err) 223 + } 224 + 225 + // Broadcast the sync event to all consumers 226 + evtCopy := *evt 227 + err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 228 + RepoSync: &evtCopy, 229 + }) 230 + if err != nil { 231 + logger.Error("failed to broadcast sync event", "error", err) 232 + return fmt.Errorf("failed to broadcast sync event: %w", err) 233 + } 234 + 235 + return nil 236 + } 237 + 238 + func (r *Relay) processIdentityEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Identity, hostname string, hostID uint64) error { 239 + logger := r.Logger.With("did", evt.Did, "seq", evt.Seq, "host", hostname, "eventType", "identity") 240 + logger.Info("relay got identity event") 241 + 242 + did, err := syntax.ParseDID(evt.Did) 243 + if err != nil { 244 + return fmt.Errorf("invalid DID in message: %w", err) 245 + } 246 + 247 + // Flush any cached DID documents for this user 248 + r.dir.Purge(ctx, did.AtIdentifier()) 249 + if err != nil { 250 + logger.Error("problem purging identity directory cache", "err", err) 251 + } 252 + 253 + // XXX: syncHostAccount doesn't need full Host? 254 + host, err := r.GetHost(ctx, hostID) 255 + if err != nil { 256 + return err 257 + } 258 + 259 + // Refetch the DID doc and update our cached keys and handle etc. 260 + account, err := r.syncHostAccount(ctx, did, host, nil) 261 + if err != nil { 262 + return err 263 + } 264 + 265 + // Broadcast the identity event to all consumers 266 + err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 267 + RepoIdentity: &comatproto.SyncSubscribeRepos_Identity{ 268 + Did: did.String(), 269 + Seq: evt.Seq, 270 + Time: evt.Time, 271 + Handle: evt.Handle, 272 + }, 273 + PrivUid: account.UID, 274 + }) 275 + if err != nil { 276 + logger.Error("failed to broadcast Identity event", "error", err) 277 + return fmt.Errorf("failed to broadcast Identity event: %w", err) 278 + } 279 + 280 + return nil 281 + } 282 + 283 + func (r *Relay) processAccountEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Account, hostname string, hostID uint64) error { 284 + logger := r.Logger.With("did", evt.Did, "seq", evt.Seq, "host", hostname, "eventType", "account") 285 + 286 + ctx, span := tracer.Start(ctx, "processAccountEvent") 287 + defer span.End() 288 + span.SetAttributes( 289 + attribute.String("did", evt.Did), 290 + attribute.Int64("seq", evt.Seq), 291 + attribute.Bool("active", evt.Active), 292 + ) 293 + 294 + did, err := syntax.ParseDID(evt.Did) 295 + if err != nil { 296 + return fmt.Errorf("invalid DID in message: %w", err) 297 + } 298 + 299 + if evt.Status != nil { 300 + span.SetAttributes(attribute.String("repo_status", *evt.Status)) 301 + } 302 + logger.Info("relay got account event") 303 + 304 + if !evt.Active && evt.Status == nil { 305 + // TODO: semantics here aren't really clear 306 + logger.Warn("dropping invalid account event", "active", evt.Active, "status", evt.Status) 307 + accountVerifyWarnings.WithLabelValues(hostname, "nostat").Inc() 308 + return nil 309 + } 310 + 311 + // Flush any cached DID documents for this user 312 + r.dir.Purge(ctx, did.AtIdentifier()) 313 + if err != nil { 314 + logger.Error("problem purging identity directory cache", "err", err) 315 + } 316 + 317 + // XXX: shouldn't need full host? 318 + host, err := r.GetHost(ctx, hostID) 319 + if err != nil { 320 + return err 321 + } 322 + 323 + // Refetch the DID doc to make sure the Host is still authoritative 324 + account, err := r.syncHostAccount(ctx, did, host, nil) 325 + if err != nil { 326 + span.RecordError(err) 327 + return err 328 + } 329 + 330 + // Check if the Host is still authoritative 331 + // if not we don't want to be propagating this account event 332 + // XXX: lock 333 + if account.HostID != hostID && !r.Config.SkipAccountHostCheck { 334 + logger.Error("account event from non-authoritative pds", 335 + "event_from", hostname, 336 + "did_doc_declared_pds", account.HostID, 337 + "account_evt", evt, 338 + ) 339 + return fmt.Errorf("event from non-authoritative pds") 340 + } 341 + 342 + // Process the account status change 343 + repoStatus := models.AccountStatusActive 344 + if !evt.Active && evt.Status != nil { 345 + repoStatus = models.AccountStatus(*evt.Status) 346 + } 347 + 348 + // XXX: lock, and parse 349 + account.UpstreamStatus = models.AccountStatus(repoStatus) 350 + err = r.db.Save(account).Error 351 + if err != nil { 352 + span.RecordError(err) 353 + return fmt.Errorf("failed to update account status: %w", err) 354 + } 355 + 356 + shouldBeActive := evt.Active 357 + status := evt.Status 358 + 359 + // override with local status 360 + // XXX: lock 361 + if account.Status == "takendown" { 362 + shouldBeActive = false 363 + s := string(models.AccountStatusTakendown) 364 + status = &s 365 + } 366 + 367 + // Broadcast the account event to all consumers 368 + err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{ 369 + RepoAccount: &comatproto.SyncSubscribeRepos_Account{ 370 + Active: shouldBeActive, 371 + Did: evt.Did, 372 + Seq: evt.Seq, 373 + Status: status, 374 + Time: evt.Time, 375 + }, 376 + PrivUid: account.UID, 377 + }) 378 + if err != nil { 379 + logger.Error("failed to broadcast Account event", "error", err) 380 + return fmt.Errorf("failed to broadcast Account event: %w", err) 381 + } 382 + 383 + return nil 384 + }
+2
cmd/relayered/relay/metrics.go
··· 31 31 Help: "The total number of sync events received", 32 32 }, []string{"pds"}) 33 33 34 + /* XXX 34 35 var repoCommitsResultCounter = promauto.NewCounterVec(prometheus.CounterOpts{ 35 36 Name: "repo_commits_result_counter", 36 37 Help: "The results of commit events received", 37 38 }, []string{"pds", "status"}) 39 + */ 38 40 39 41 var eventsSentCounter = promauto.NewCounterVec(prometheus.CounterOpts{ 40 42 Name: "events_sent_counter",
+3 -3
cmd/relayered/relay/relay.go
··· 26 26 HostChecker HostChecker 27 27 Config RelayConfig 28 28 29 - // extUserLk serializes a section of syncHostAccount() 29 + // accountLk serializes a section of syncHostAccount() 30 30 // TODO: at some point we will want to lock specific DIDs, this lock as is 31 31 // is overly broad, but i dont expect it to be a bottleneck for now 32 - extUserLk sync.Mutex 32 + accountLk sync.Mutex 33 33 34 34 // Management of Socket Consumers 35 35 consumersLk sync.RWMutex ··· 96 96 slOpts.DefaultRepoLimit = config.DefaultRepoLimit 97 97 slOpts.ConcurrencyPerHost = config.ConcurrencyPerHost 98 98 slOpts.MaxQueuePerHost = config.MaxQueuePerHost 99 - s, err := NewSlurper(db, r.handleFedEvent, slOpts, r.Logger) 99 + s, err := NewSlurper(db, r.processRepoEvent, slOpts, r.Logger) 100 100 if err != nil { 101 101 return nil, err 102 102 }
+96 -102
cmd/relayered/relay/slurper.go
··· 19 19 "gorm.io/gorm" 20 20 ) 21 21 22 - type ProcessMessageFunc func(context.Context, *models.Host, *stream.XRPCStreamEvent) error 22 + type ProcessMessageFunc func(ctx context.Context, evt *stream.XRPCStreamEvent, hostname string, hostID uint64) error 23 23 24 24 type Slurper struct { 25 25 cb ProcessMessageFunc ··· 37 37 shutdownChan chan bool 38 38 shutdownResult chan []error 39 39 40 - log *slog.Logger 40 + logger *slog.Logger 41 41 } 42 42 43 43 type Limiters struct { ··· 71 71 72 72 // represents an active client connection to a remote host 73 73 type Subscription struct { 74 - Host *models.Host 74 + Hostname string 75 + HostID uint64 75 76 LastSeq int64 // XXX: switch to an atomic 76 77 Limiters *Limiters 77 78 ··· 83 84 func (sub *Subscription) UpdateSeq(seq int64) { 84 85 sub.lk.Lock() 85 86 defer sub.lk.Unlock() 86 - sub.Host.LastSeq = seq 87 + sub.LastSeq = seq 87 88 } 88 89 89 90 func NewSlurper(db *gorm.DB, cb ProcessMessageFunc, config *SlurperConfig, logger *slog.Logger) (*Slurper, error) { ··· 106 107 shutdownChan: make(chan bool), 107 108 shutdownResult: make(chan []error), 108 109 NewHostPerDayLimiter: newHostPerDayLimiter, 109 - log: logger, 110 + logger: logger, 110 111 } 111 112 112 113 // Start a goroutine to flush cursors to the DB every 30s ··· 114 115 for { 115 116 select { 116 117 case <-s.shutdownChan: 117 - s.log.Info("flushing Host cursors on shutdown") 118 + s.logger.Info("flushing Host cursors on shutdown") 118 119 ctx := context.Background() 119 120 var errs []error 120 121 if errs = s.flushCursors(ctx); len(errs) > 0 { 121 122 for _, err := range errs { 122 - s.log.Error("failed to flush cursors on shutdown", "err", err) 123 + s.logger.Error("failed to flush cursors on shutdown", "err", err) 123 124 } 124 125 } 125 - s.log.Info("done flushing Host cursors on shutdown") 126 + s.logger.Info("done flushing Host cursors on shutdown") 126 127 s.shutdownResult <- errs 127 128 return 128 129 case <-time.After(time.Second * 10): 129 - s.log.Debug("flushing Host cursors") 130 + s.logger.Debug("flushing Host cursors") 130 131 ctx := context.Background() 131 132 if errs := s.flushCursors(ctx); len(errs) > 0 { 132 133 for _, err := range errs { 133 - s.log.Error("failed to flush cursors", "err", err) 134 + s.logger.Error("failed to flush cursors", "err", err) 134 135 } 135 136 } 136 - s.log.Debug("done flushing Host cursors") 137 + s.logger.Debug("done flushing Host cursors") 137 138 } 138 139 } 139 140 }() ··· 216 217 // Shutdown shuts down the slurper 217 218 func (s *Slurper) Shutdown() []error { 218 219 s.shutdownChan <- true 219 - s.log.Info("waiting for slurper shutdown") 220 + s.logger.Info("waiting for slurper shutdown") 220 221 errs := <-s.shutdownResult 221 222 if len(errs) > 0 { 222 223 for _, err := range errs { 223 - s.log.Error("shutdown error", "err", err) 224 + s.logger.Error("shutdown error", "err", err) 224 225 } 225 226 } 226 - s.log.Info("slurper shutdown complete") 227 + s.logger.Info("slurper shutdown complete") 227 228 return errs 228 229 } 229 230 ··· 242 243 243 244 ctx, cancel := context.WithCancel(context.Background()) 244 245 sub := Subscription{ 245 - Host: host, 246 - ctx: ctx, 247 - cancel: cancel, 246 + Hostname: host.Hostname, 247 + HostID: host.ID, 248 + ctx: ctx, 249 + cancel: cancel, 248 250 } 249 251 s.active[host.Hostname] = &sub 250 252 ··· 292 294 if newHost { 293 295 u = fmt.Sprintf("%s?cursor=%d", u, cursor) 294 296 } 295 - con, res, err := d.DialContext(ctx, u, nil) 297 + conn, res, err := d.DialContext(ctx, u, nil) 296 298 if err != nil { 297 - s.log.Warn("dialing failed", "host", host.Hostname, "err", err, "backoff", backoff) 299 + s.logger.Warn("dialing failed", "host", host.Hostname, "err", err, "backoff", backoff) 298 300 time.Sleep(sleepForBackoff(backoff)) 299 301 backoff++ 300 302 301 303 if backoff > 15 { 302 - s.log.Warn("host does not appear to be online, disabling for now", "host", host.Hostname) 304 + s.logger.Warn("host does not appear to be online, disabling for now", "host", host.Hostname) 303 305 if err := s.db.Model(&models.Host{}).Where("id = ?", host.ID).Update("registered", false).Error; err != nil { 304 - s.log.Error("failed to unregister failing host", "err", err) 306 + s.logger.Error("failed to unregister failing host", "err", err) 305 307 } 306 308 307 309 return ··· 310 312 continue 311 313 } 312 314 313 - s.log.Info("event subscription response", "code", res.StatusCode, "url", u) 315 + s.logger.Info("event subscription response", "code", res.StatusCode, "url", u) 314 316 315 317 curCursor := cursor 316 - if err := s.handleConnection(ctx, host, con, &cursor, sub); err != nil { 318 + if err := s.handleConnection(ctx, conn, &cursor, sub); err != nil { 317 319 if errors.Is(err, ErrTimeoutShutdown) { 318 - s.log.Info("shutting down host subscription after timeout", "host", host.Hostname, "time", EventsTimeout) 320 + s.logger.Info("shutting down host subscription after timeout", "host", host.Hostname, "time", EventsTimeout) 319 321 return 320 322 } 321 - s.log.Warn("connection to failed", "host", host.Hostname, "err", err) 323 + s.logger.Warn("connection to failed", "host", host.Hostname, "err", err) 322 324 // TODO: measure the last N connection error times and if they're coming too fast reconnect slower or don't reconnect and wait for requestCrawl 323 325 } 324 326 ··· 342 344 343 345 var EventsTimeout = time.Minute 344 346 345 - func (s *Slurper) handleConnection(ctx context.Context, host *models.Host, con *websocket.Conn, lastCursor *int64, sub *Subscription) error { 347 + func (s *Slurper) handleConnection(ctx context.Context, conn *websocket.Conn, lastCursor *int64, sub *Subscription) error { 346 348 ctx, cancel := context.WithCancel(ctx) 347 349 defer cancel() 348 350 349 351 rsc := &stream.RepoStreamCallbacks{ 350 352 RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error { 351 - s.log.Debug("got remote repo event", "host", host.Hostname, "repo", evt.Repo, "seq", evt.Seq) 352 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 353 - RepoCommit: evt, 354 - }); err != nil { 355 - s.log.Error("failed handling event", "host", host.Hostname, "seq", evt.Seq, "err", err) 353 + logger := s.logger.With("host", sub.Hostname, "did", evt.Repo, "seq", evt.Seq, "eventType", "commit") 354 + logger.Debug("got remote repo event") 355 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoCommit: evt}, sub.Hostname, sub.HostID); err != nil { 356 + logger.Error("failed handling event", "err", err) 356 357 } 357 358 *lastCursor = evt.Seq 358 359 ··· 361 362 return nil 362 363 }, 363 364 RepoSync: func(evt *comatproto.SyncSubscribeRepos_Sync) error { 364 - s.log.Debug("got remote repo event", "host", host.Hostname, "repo", evt.Did, "seq", evt.Seq) 365 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 366 - RepoSync: evt, 367 - }); err != nil { 368 - s.log.Error("failed handling event", "host", host.Hostname, "seq", evt.Seq, "err", err) 365 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "sync") 366 + logger.Debug("got remote repo event") 367 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoSync: evt}, sub.Hostname, sub.HostID); err != nil { 368 + s.logger.Error("failed handling event", "err", err) 369 369 } 370 370 *lastCursor = evt.Seq 371 371 ··· 373 373 374 374 return nil 375 375 }, 376 - RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error { 377 - s.log.Debug("got remote handle update event", "host", host.Hostname, "did", evt.Did, "handle", evt.Handle) 378 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 379 - RepoHandle: evt, 380 - }); err != nil { 381 - s.log.Error("failed handling event", "host", host.Hostname, "seq", evt.Seq, "err", err) 376 + RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error { 377 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "identity") 378 + logger.Debug("identity event") 379 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoIdentity: evt}, sub.Hostname, sub.HostID); err != nil { 380 + logger.Error("failed handling event", "err", err) 382 381 } 383 382 *lastCursor = evt.Seq 384 383 ··· 386 385 387 386 return nil 388 387 }, 389 - RepoMigrate: func(evt *comatproto.SyncSubscribeRepos_Migrate) error { 390 - s.log.Debug("got remote repo migrate event", "host", host.Hostname, "did", evt.Did, "migrateTo", evt.MigrateTo) 391 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 392 - RepoMigrate: evt, 393 - }); err != nil { 394 - s.log.Error("failed handling event", "host", host.Hostname, "seq", evt.Seq, "err", err) 388 + RepoAccount: func(evt *comatproto.SyncSubscribeRepos_Account) error { 389 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "account") 390 + s.logger.Debug("account event") 391 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoAccount: evt}, sub.Hostname, sub.HostID); err != nil { 392 + logger.Error("failed handling event", "err", err) 395 393 } 396 394 *lastCursor = evt.Seq 397 395 ··· 399 397 400 398 return nil 401 399 }, 402 - RepoTombstone: func(evt *comatproto.SyncSubscribeRepos_Tombstone) error { 403 - s.log.Debug("got remote repo tombstone event", "host", host.Hostname, "did", evt.Did) 404 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 405 - RepoTombstone: evt, 406 - }); err != nil { 407 - s.log.Error("failed handling event", "host", host.Hostname, "seq", evt.Seq, "err", err) 400 + Error: func(evt *stream.ErrorFrame) error { 401 + switch evt.Error { 402 + case "FutureCursor": 403 + // XXX: need test coverage for this path 404 + // if we get a FutureCursor frame, reset our sequence number for this host 405 + if err := s.db.Table("host").Where("id = ?", sub.HostID).Update("last_seq", 0).Error; err != nil { 406 + return err 407 + } 408 + 409 + *lastCursor = 0 410 + return fmt.Errorf("got FutureCursor frame, reset cursor tracking for host") 411 + default: 412 + return fmt.Errorf("error frame: %s: %s", evt.Error, evt.Message) 413 + } 414 + }, 415 + RepoInfo: func(info *comatproto.SyncSubscribeRepos_Info) error { 416 + s.logger.Debug("info event", "name", info.Name, "message", info.Message, "host", sub.Hostname) 417 + return nil 418 + }, 419 + RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error { // DEPRECATED 420 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "handle") 421 + logger.Debug("got remote handle update event", "handle", evt.Handle) 422 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoHandle: evt}, sub.Hostname, sub.HostID); err != nil { 423 + logger.Error("failed handling event", "err", err) 408 424 } 409 425 *lastCursor = evt.Seq 410 426 ··· 412 428 413 429 return nil 414 430 }, 415 - RepoInfo: func(info *comatproto.SyncSubscribeRepos_Info) error { 416 - s.log.Debug("info event", "name", info.Name, "message", info.Message, "host", host.Hostname) 417 - return nil 418 - }, 419 - RepoIdentity: func(ident *comatproto.SyncSubscribeRepos_Identity) error { 420 - s.log.Debug("identity event", "did", ident.Did) 421 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 422 - RepoIdentity: ident, 423 - }); err != nil { 424 - s.log.Error("failed handling event", "host", host.Hostname, "seq", ident.Seq, "err", err) 431 + RepoMigrate: func(evt *comatproto.SyncSubscribeRepos_Migrate) error { // DEPRECATED 432 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "migrate") 433 + logger.Debug("got remote repo migrate event", "migrateTo", evt.MigrateTo) 434 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoMigrate: evt}, sub.Hostname, sub.HostID); err != nil { 435 + logger.Error("failed handling event", "err", err) 425 436 } 426 - *lastCursor = ident.Seq 437 + *lastCursor = evt.Seq 427 438 428 439 sub.UpdateSeq(*lastCursor) 429 440 430 441 return nil 431 442 }, 432 - RepoAccount: func(acct *comatproto.SyncSubscribeRepos_Account) error { 433 - s.log.Debug("account event", "did", acct.Did, "status", acct.Status) 434 - if err := s.cb(context.TODO(), host, &stream.XRPCStreamEvent{ 435 - RepoAccount: acct, 436 - }); err != nil { 437 - s.log.Error("failed handling event", "host", host.Hostname, "seq", acct.Seq, "err", err) 443 + RepoTombstone: func(evt *comatproto.SyncSubscribeRepos_Tombstone) error { // DEPRECATED 444 + logger := s.logger.With("host", sub.Hostname, "did", evt.Did, "seq", evt.Seq, "eventType", "tombstone") 445 + logger.Debug("got remote repo tombstone event") 446 + if err := s.cb(context.TODO(), &stream.XRPCStreamEvent{RepoTombstone: evt}, sub.Hostname, sub.HostID); err != nil { 447 + logger.Error("failed handling event", "err", err) 438 448 } 439 - *lastCursor = acct.Seq 449 + *lastCursor = evt.Seq 440 450 441 451 sub.UpdateSeq(*lastCursor) 442 452 443 453 return nil 444 454 }, 445 - // TODO: all the other event types (handle change, migration, etc) 446 - Error: func(errf *stream.ErrorFrame) error { 447 - switch errf.Error { 448 - case "FutureCursor": 449 - // XXX: need test coverage for this path 450 - // if we get a FutureCursor frame, reset our sequence number for this host 451 - if err := s.db.Table("host").Where("id = ?", host.ID).Update("last_seq", 0).Error; err != nil { 452 - return err 453 - } 454 - 455 - *lastCursor = 0 456 - return fmt.Errorf("got FutureCursor frame, reset cursor tracking for host") 457 - default: 458 - return fmt.Errorf("error frame: %s: %s", errf.Error, errf.Message) 459 - } 460 - }, 461 455 } 462 456 463 - lims := s.GetOrCreateLimiters(host.ID) 457 + lims := s.GetOrCreateLimiters(sub.HostID) 464 458 465 459 limiters := []*slidingwindow.Limiter{ 466 460 lims.PerSecond, ··· 473 467 pool := parallel.NewScheduler( 474 468 100, 475 469 1_000, 476 - con.RemoteAddr().String(), 470 + conn.RemoteAddr().String(), 477 471 instrumentedRSC.EventHandler, 478 472 ) 479 - return stream.HandleRepoStream(ctx, con, pool, nil) 473 + return stream.HandleRepoStream(ctx, conn, pool, nil) 480 474 } 481 475 482 476 type cursorSnapshot struct { 483 - id uint64 477 + hostID uint64 484 478 cursor int64 485 479 } 486 480 ··· 497 491 for _, sub := range s.active { 498 492 sub.lk.RLock() 499 493 cursors = append(cursors, cursorSnapshot{ 500 - id: sub.Host.ID, 501 - cursor: sub.Host.LastSeq, 494 + hostID: sub.HostID, 495 + cursor: sub.LastSeq, 502 496 }) 503 497 sub.lk.RUnlock() 504 498 } ··· 509 503 510 504 tx := s.db.WithContext(ctx).Begin() 511 505 for _, cursor := range cursors { 512 - if err := tx.WithContext(ctx).Model(models.Host{}).Where("id = ?", cursor.id).UpdateColumn("last_seq", cursor.cursor).Error; err != nil { 506 + if err := tx.WithContext(ctx).Model(models.Host{}).Where("id = ?", cursor.hostID).UpdateColumn("last_seq", cursor.cursor).Error; err != nil { 513 507 errs = append(errs, err) 514 508 } else { 515 509 okcount++ ··· 519 513 errs = append(errs, err) 520 514 } 521 515 dt := time.Since(start) 522 - s.log.Info("flushCursors", "dt", dt, "ok", okcount, "errs", len(errs)) 516 + s.logger.Info("flushCursors", "dt", dt, "ok", okcount, "errs", len(errs)) 523 517 524 518 return errs 525 519 } ··· 535 529 return out 536 530 } 537 531 538 - func (s *Slurper) KillUpstreamConnection(host string, block bool) error { 532 + func (s *Slurper) KillUpstreamConnection(hostname string, ban bool) error { 539 533 s.lk.Lock() 540 534 defer s.lk.Unlock() 541 535 542 - ac, ok := s.active[host] 536 + ac, ok := s.active[hostname] 543 537 if !ok { 544 - return fmt.Errorf("killing connection %q: %w", host, ErrNoActiveConnection) 538 + return fmt.Errorf("killing connection %q: %w", hostname, ErrNoActiveConnection) 545 539 } 546 540 ac.cancel() 547 541 // cleanup in the run thread subscribeWithRedialer() will delete(s.active, host) 548 542 549 - if block { 550 - if err := s.db.Model(models.Host{}).Where("id = ?", ac.Host.ID).UpdateColumn("blocked", true).Error; err != nil { 551 - return fmt.Errorf("failed to set host as blocked: %w", err) 543 + if ban { 544 + if err := s.db.Model(models.Host{}).Where("id = ?", ac.HostID).UpdateColumn("status", "banned").Error; err != nil { 545 + return fmt.Errorf("failed to set host as banned: %w", err) 552 546 } 553 547 } 554 548
-217
cmd/relayered/relay/subscribe_repos.go
··· 1 - package relay 2 - 3 - import ( 4 - "context" 5 - "fmt" 6 - "net" 7 - "net/http" 8 - "sync" 9 - "time" 10 - 11 - "github.com/bluesky-social/indigo/cmd/relayered/stream" 12 - 13 - "github.com/gorilla/websocket" 14 - promclient "github.com/prometheus/client_golang/prometheus" 15 - dto "github.com/prometheus/client_model/go" 16 - ) 17 - 18 - type SocketConsumer struct { 19 - UserAgent string 20 - RemoteAddr string 21 - ConnectedAt time.Time 22 - EventsSent promclient.Counter 23 - } 24 - 25 - func (r *Relay) registerConsumer(c *SocketConsumer) uint64 { 26 - r.consumersLk.Lock() 27 - defer r.consumersLk.Unlock() 28 - 29 - id := r.nextConsumerID 30 - r.nextConsumerID++ 31 - 32 - r.consumers[id] = c 33 - 34 - return id 35 - } 36 - 37 - func (r *Relay) cleanupConsumer(id uint64) { 38 - r.consumersLk.Lock() 39 - defer r.consumersLk.Unlock() 40 - 41 - c := r.consumers[id] 42 - 43 - var m = &dto.Metric{} 44 - if err := c.EventsSent.Write(m); err != nil { 45 - r.Logger.Error("failed to get sent counter", "err", err) 46 - } 47 - 48 - r.Logger.Info("consumer disconnected", 49 - "consumer_id", id, 50 - "remote_addr", c.RemoteAddr, 51 - "user_agent", c.UserAgent, 52 - "events_sent", m.Counter.GetValue()) 53 - 54 - delete(r.consumers, id) 55 - } 56 - 57 - // Main HTTP request handler for clients connecting to the firehose (com.atproto.sync.subscribeRepos) 58 - func (r *Relay) HandleSubscribeRepos(resp http.ResponseWriter, req *http.Request, since *int64, realIP string) error { 59 - 60 - ctx, cancel := context.WithCancel(req.Context()) 61 - defer cancel() 62 - 63 - conn, err := websocket.Upgrade(resp, req, resp.Header(), 10<<10, 10<<10) 64 - if err != nil { 65 - return fmt.Errorf("upgrading websocket: %w", err) 66 - } 67 - 68 - defer conn.Close() 69 - 70 - lastWriteLk := sync.Mutex{} 71 - lastWrite := time.Now() 72 - 73 - // Start a goroutine to ping the client every 30 seconds to check if it's 74 - // still alive. If the client doesn't respond to a ping within 5 seconds, 75 - // we'll close the connection and teardown the consumer. 76 - go func() { 77 - ticker := time.NewTicker(30 * time.Second) 78 - defer ticker.Stop() 79 - 80 - for { 81 - select { 82 - case <-ticker.C: 83 - lastWriteLk.Lock() 84 - lw := lastWrite 85 - lastWriteLk.Unlock() 86 - 87 - if time.Since(lw) < 30*time.Second { 88 - continue 89 - } 90 - 91 - if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil { 92 - r.Logger.Warn("failed to ping client", "err", err) 93 - cancel() 94 - return 95 - } 96 - case <-ctx.Done(): 97 - return 98 - } 99 - } 100 - }() 101 - 102 - conn.SetPingHandler(func(message string) error { 103 - err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60)) 104 - if err == websocket.ErrCloseSent { 105 - return nil 106 - } else if e, ok := err.(net.Error); ok && e.Temporary() { 107 - return nil 108 - } 109 - return err 110 - }) 111 - 112 - // Start a goroutine to read messages from the client and discard them. 113 - go func() { 114 - for { 115 - _, _, err := conn.ReadMessage() 116 - if err != nil { 117 - r.Logger.Warn("failed to read message from client", "err", err) 118 - cancel() 119 - return 120 - } 121 - } 122 - }() 123 - 124 - ident := realIP + "-" + req.UserAgent() 125 - 126 - evts, cleanup, err := r.Events.Subscribe(ctx, ident, func(evt *stream.XRPCStreamEvent) bool { return true }, since) 127 - if err != nil { 128 - return err 129 - } 130 - defer cleanup() 131 - 132 - // Keep track of the consumer for metrics and admin endpoints 133 - consumer := SocketConsumer{ 134 - RemoteAddr: realIP, 135 - UserAgent: req.UserAgent(), 136 - ConnectedAt: time.Now(), 137 - } 138 - sentCounter := eventsSentCounter.WithLabelValues(consumer.RemoteAddr, consumer.UserAgent) 139 - consumer.EventsSent = sentCounter 140 - 141 - consumerID := r.registerConsumer(&consumer) 142 - defer r.cleanupConsumer(consumerID) 143 - 144 - logger := r.Logger.With( 145 - "consumer_id", consumerID, 146 - "remote_addr", consumer.RemoteAddr, 147 - "user_agent", consumer.UserAgent, 148 - ) 149 - 150 - logger.Info("new consumer", "cursor", since) 151 - 152 - for { 153 - select { 154 - case evt, ok := <-evts: 155 - if !ok { 156 - logger.Error("event stream closed unexpectedly") 157 - return nil 158 - } 159 - 160 - wc, err := conn.NextWriter(websocket.BinaryMessage) 161 - if err != nil { 162 - logger.Error("failed to get next writer", "err", err) 163 - return err 164 - } 165 - 166 - if evt.Preserialized != nil { 167 - _, err = wc.Write(evt.Preserialized) 168 - } else { 169 - err = evt.Serialize(wc) 170 - } 171 - if err != nil { 172 - return fmt.Errorf("failed to write event: %w", err) 173 - } 174 - 175 - if err := wc.Close(); err != nil { 176 - logger.Warn("failed to flush-close our event write", "err", err) 177 - return nil 178 - } 179 - 180 - lastWriteLk.Lock() 181 - lastWrite = time.Now() 182 - lastWriteLk.Unlock() 183 - sentCounter.Inc() 184 - case <-ctx.Done(): 185 - return nil 186 - } 187 - } 188 - } 189 - 190 - type ConsumerInfo struct { 191 - ID uint64 `json:"id"` 192 - RemoteAddr string `json:"remote_addr"` 193 - UserAgent string `json:"user_agent"` 194 - EventsConsumed uint64 `json:"events_consumed"` 195 - ConnectedAt time.Time `json:"connected_at"` 196 - } 197 - 198 - func (r *Relay) ListConsumers() []ConsumerInfo { 199 - r.consumersLk.RLock() 200 - defer r.consumersLk.RUnlock() 201 - 202 - info := make([]ConsumerInfo, 0, len(r.consumers)) 203 - for id, c := range r.consumers { 204 - var m = &dto.Metric{} 205 - if err := c.EventsSent.Write(m); err != nil { 206 - continue 207 - } 208 - info = append(info, ConsumerInfo{ 209 - ID: id, 210 - RemoteAddr: c.RemoteAddr, 211 - UserAgent: c.UserAgent, 212 - EventsConsumed: uint64(m.Counter.GetValue()), 213 - ConnectedAt: c.ConnectedAt, 214 - }) 215 - } 216 - return info 217 - }
+10 -12
cmd/relayered/relay/validator.go
··· 58 58 } 59 59 60 60 type NextCommitHandler interface { 61 - HandleCommit(ctx context.Context, host *models.Host, uid uint64, did string, commit *comatproto.SyncSubscribeRepos_Commit) error 61 + HandleCommit(ctx context.Context, hostname string, uid uint64, did string, commit *comatproto.SyncSubscribeRepos_Commit) error 62 62 } 63 63 64 64 type userLock struct { ··· 99 99 } 100 100 } 101 101 102 - func (val *Validator) HandleCommit(ctx context.Context, host *models.Host, account *models.Account, commit *comatproto.SyncSubscribeRepos_Commit, prevRev *syntax.TID, prevData *cid.Cid) (commitDataCID *cid.Cid, err error) { 102 + func (val *Validator) HandleCommit(ctx context.Context, hostname string, account *models.Account, commit *comatproto.SyncSubscribeRepos_Commit, prevRev *syntax.TID, prevData *cid.Cid) (commitDataCID *cid.Cid, err error) { 103 103 uid := account.UID 104 104 unlock := val.lockUser(ctx, uid) 105 105 defer unlock() 106 - repoFragment, err := val.VerifyCommitMessage(ctx, host, commit, prevRev, prevData) 106 + repoFragment, err := val.VerifyCommitMessage(ctx, hostname, commit, prevRev, prevData) 107 107 if err != nil { 108 108 return nil, err 109 109 } ··· 124 124 125 125 var ErrNewRevBeforePrevRev = &revOutOfOrderError{} 126 126 127 - func (val *Validator) VerifyCommitMessage(ctx context.Context, host *models.Host, msg *comatproto.SyncSubscribeRepos_Commit, prevRev *syntax.TID, prevData *cid.Cid) (*repo.Repo, error) { 128 - hostname := host.Hostname 127 + func (val *Validator) VerifyCommitMessage(ctx context.Context, hostname string, msg *comatproto.SyncSubscribeRepos_Commit, prevRev *syntax.TID, prevData *cid.Cid) (*repo.Repo, error) { 129 128 hasWarning := false 130 129 commitVerifyStarts.Inc() 131 130 logger := slog.Default().With("did", msg.Repo, "rev", msg.Rev, "seq", msg.Seq, "time", msg.Time) ··· 163 162 //logger.Warn("event with tooBig flag set") 164 163 commitVerifyWarnings.WithLabelValues(hostname, "big").Inc() 165 164 // XXX: induction trace log 166 - val.log.Warn("commit tooBig", "seq", msg.Seq, "host", host.Hostname, "repo", msg.Repo) 165 + val.log.Warn("commit tooBig", "seq", msg.Seq, "host", hostname, "repo", msg.Repo) 167 166 hasWarning = true 168 167 } 169 168 if msg.Rebase { 170 169 //logger.Warn("event with rebase flag set") 171 170 commitVerifyWarnings.WithLabelValues(hostname, "reb").Inc() 172 171 // XXX: induction trace log 173 - val.log.Warn("commit rebase", "seq", msg.Seq, "host", host.Hostname, "repo", msg.Repo) 172 + val.log.Warn("commit rebase", "seq", msg.Seq, "host", hostname, "repo", msg.Repo) 174 173 hasWarning = true 175 174 } 176 175 ··· 228 227 if o.Prev == nil { 229 228 logger.Debug("can't invert legacy op", "action", o.Action) 230 229 // XXX: induction trace log 231 - val.log.Warn("commit delete op", "seq", msg.Seq, "host", host.Hostname, "repo", msg.Repo) 230 + val.log.Warn("commit delete op", "seq", msg.Seq, "host", hostname, "repo", msg.Repo) 232 231 commitVerifyOkish.WithLabelValues(hostname, "del").Inc() 233 232 return repoFragment, nil 234 233 } ··· 236 235 if o.Prev == nil { 237 236 logger.Debug("can't invert legacy op", "action", o.Action) 238 237 // XXX: induction trace log 239 - val.log.Warn("commit update op", "seq", msg.Seq, "host", host.Hostname, "repo", msg.Repo) 238 + val.log.Warn("commit update op", "seq", msg.Seq, "host", hostname, "repo", msg.Repo) 240 239 commitVerifyOkish.WithLabelValues(hostname, "up").Inc() 241 240 return repoFragment, nil 242 241 } ··· 249 248 if *c != *prevData { 250 249 commitVerifyWarnings.WithLabelValues(hostname, "pr").Inc() 251 250 // XXX: induction trace log 252 - val.log.Warn("commit prevData mismatch", "seq", msg.Seq, "host", host.Hostname, "repo", msg.Repo) 251 + val.log.Warn("commit prevData mismatch", "seq", msg.Seq, "host", hostname, "repo", msg.Repo) 253 252 hasWarning = true 254 253 } 255 254 } else { ··· 305 304 } 306 305 307 306 // HandleSync checks signed commit from a #sync message 308 - func (val *Validator) HandleSync(ctx context.Context, host *models.Host, msg *comatproto.SyncSubscribeRepos_Sync) (commitCID, commitDataCID *cid.Cid, err error) { 309 - hostname := host.Hostname 307 + func (val *Validator) HandleSync(ctx context.Context, hostname string, msg *comatproto.SyncSubscribeRepos_Sync) (commitCID, commitDataCID *cid.Cid, err error) { 310 308 hasWarning := false 311 309 312 310 did, err := syntax.ParseDID(msg.Did)