+1 -2

knotmirror/config/config.go

Reviewed

··· 33 33 } 34 34 35 35 type SlurperConfig struct { 36 36 - PersistCursorPeriod time.Duration `env:"PERSIST_CURSOR_PERIOD, default=4s"` 37 37 - ConcurrencyPerHost int `env:"CONCURRENCY, default=4"` 36 36 + ConcurrencyPerHost int `env:"CONCURRENCY, default=4"` 38 37 } 39 38 40 39 func Load(ctx context.Context) (*Config, error) {

+1 -1

knotmirror/db/db.go

Reviewed

··· 62 62 hostname text not null, 63 63 no_ssl boolean not null default false, 64 64 status text not null default 'active', 65 65 - last_seq bigint not null default -1, 65 65 + last_tid text not null default '', 66 66 db_created_at timestamptz not null default now(), 67 67 db_updated_at timestamptz not null default now(), 68 68

+7 -29

knotmirror/db/hosts.go

Reviewed

··· 5 5 "database/sql" 6 6 "errors" 7 7 "fmt" 8 8 - "log" 9 8 10 9 "tangled.org/core/knotmirror/models" 11 10 ) 12 11 13 12 func UpsertHost(ctx context.Context, e *sql.DB, host *models.Host) error { 14 13 if _, err := e.ExecContext(ctx, 15 15 - `insert into hosts (hostname, no_ssl, status, last_seq) 14 14 + `insert into hosts (hostname, no_ssl, status, last_tid) 16 15 values ($1, $2, $3, $4) 17 16 on conflict(hostname) do update set 18 17 no_ssl = excluded.no_ssl, 19 18 status = excluded.status, 20 20 - last_seq = excluded.last_seq 19 19 + last_tid = excluded.last_tid 21 20 `, 22 21 host.Hostname, 23 22 host.NoSSL, 24 23 host.Status, 25 25 - host.LastSeq, 24 24 + host.LastTid, 26 25 ); err != nil { 27 26 return fmt.Errorf("upserting host: %w", err) 28 27 } ··· 32 31 func GetHost(ctx context.Context, e *sql.DB, hostname string) (*models.Host, error) { 33 32 var host models.Host 34 33 if err := e.QueryRowContext(ctx, 35 35 - `select hostname, no_ssl, status, last_seq 34 34 + `select hostname, no_ssl, status, last_tid 36 35 from hosts where hostname = $1`, 37 36 hostname, 38 37 ).Scan( 39 38 &host.Hostname, 40 39 &host.NoSSL, 41 40 &host.Status, 42 42 - &host.LastSeq, 41 41 + &host.LastTid, 43 42 ); err != nil { 44 43 if errors.Is(err, sql.ErrNoRows) { 45 44 return nil, nil ··· 49 48 return &host, nil 50 49 } 51 50 52 52 - func StoreCursors(ctx context.Context, e *sql.DB, cursors []models.HostCursor) error { 53 53 - tx, err := e.BeginTx(ctx, nil) 54 54 - if err != nil { 55 55 - return fmt.Errorf("starting transaction: %w", err) 56 56 - } 57 57 - defer tx.Rollback() 58 58 - for _, cur := range cursors { 59 59 - if cur.LastSeq <= 0 { 60 60 - continue 61 61 - } 62 62 - if _, err := tx.ExecContext(ctx, 63 63 - `update hosts set last_seq = $1 where hostname = $2`, 64 64 - cur.LastSeq, 65 65 - cur.Hostname, 66 66 - ); err != nil { 67 67 - log.Println("failed to persist host cursor", "host", cur.Hostname, "lastSeq", cur.LastSeq, "err", err) 68 68 - } 69 69 - } 70 70 - return tx.Commit() 71 71 - } 72 72 - 73 51 func ListHosts(ctx context.Context, e *sql.DB, status models.HostStatus) ([]models.Host, error) { 74 52 rows, err := e.QueryContext(ctx, 75 75 - `select hostname, no_ssl, status, last_seq 53 53 + `select hostname, no_ssl, status, last_tid 76 54 from hosts 77 55 where status = $1`, 78 56 status, ··· 89 67 &host.Hostname, 90 68 &host.NoSSL, 91 69 &host.Status, 92 92 - &host.LastSeq, 70 70 + &host.LastTid, 93 71 ); err != nil { 94 72 return nil, fmt.Errorf("scanning row: %w", err) 95 73 }

+78

knotmirror/db/migrations_list.go

Reviewed

··· 6 6 "fmt" 7 7 8 8 "tangled.org/core/log" 9 9 + "tangled.org/core/tid" 9 10 ) 10 11 11 12 var Migrations = []Migration{ ··· 13 14 Name: "repos_pk_to_repo_did", 14 15 Fn: reposPkToRepoDid, 15 16 }, 17 17 + { 18 18 + Name: "hosts_add_last_tid", 19 19 + Fn: hostsAddLastTid, 20 20 + }, 21 21 + { 22 22 + Name: "hosts_backfill_last_tid_from_last_seq", 23 23 + Fn: hostsBackfillLastTidFromLastSeq, 24 24 + }, 25 25 + { 26 26 + Name: "hosts_drop_last_seq", 27 27 + Fn: hostsDropLastSeq, 28 28 + }, 29 29 + } 30 30 + 31 31 + func hostsAddLastTid(ctx context.Context, tx *sql.Tx) error { 32 32 + _, err := tx.ExecContext(ctx, 33 33 + `alter table hosts add column if not exists last_tid text not null default ''`, 34 34 + ) 35 35 + return err 36 36 + } 37 37 + 38 38 + func hostsBackfillLastTidFromLastSeq(ctx context.Context, tx *sql.Tx) error { 39 39 + var hasLastSeq bool 40 40 + if err := tx.QueryRowContext(ctx, 41 41 + `select exists ( 42 42 + select 1 from information_schema.columns 43 43 + where table_name = 'hosts' and column_name = 'last_seq' 44 44 + )`, 45 45 + ).Scan(&hasLastSeq); err != nil { 46 46 + return fmt.Errorf("checking last_seq column: %w", err) 47 47 + } 48 48 + if !hasLastSeq { 49 49 + return nil 50 50 + } 51 51 + 52 52 + rows, err := tx.QueryContext(ctx, 53 53 + `select hostname, last_seq from hosts where last_seq > 0 and last_tid = ''`, 54 54 + ) 55 55 + if err != nil { 56 56 + return fmt.Errorf("reading hosts: %w", err) 57 57 + } 58 58 + defer rows.Close() 59 59 + 60 60 + type entry struct { 61 61 + hostname string 62 62 + lastSeq int64 63 63 + } 64 64 + var entries []entry 65 65 + for rows.Next() { 66 66 + var e entry 67 67 + if err := rows.Scan(&e.hostname, &e.lastSeq); err != nil { 68 68 + return fmt.Errorf("scanning host row: %w", err) 69 69 + } 70 70 + entries = append(entries, e) 71 71 + } 72 72 + if err := rows.Err(); err != nil { 73 73 + return fmt.Errorf("iterating host rows: %w", err) 74 74 + } 75 75 + 76 76 + for _, e := range entries { 77 77 + gen := tid.MonotonicGenerator{} 78 78 + newTid := gen.FromNanos(e.lastSeq) 79 79 + if _, err := tx.ExecContext(ctx, 80 80 + `update hosts set last_tid = $1 where hostname = $2`, 81 81 + newTid, e.hostname, 82 82 + ); err != nil { 83 83 + return fmt.Errorf("backfilling %s: %w", e.hostname, err) 84 84 + } 85 85 + } 86 86 + return nil 87 87 + } 88 88 + 89 89 + func hostsDropLastSeq(ctx context.Context, tx *sql.Tx) error { 90 90 + _, err := tx.ExecContext(ctx, 91 91 + `alter table hosts drop column if exists last_seq`, 92 92 + ) 93 93 + return err 16 94 } 17 95 18 96 func reposPkToRepoDid(ctx context.Context, tx *sql.Tx) error {

+44

knotmirror/knotstream/hosts_cursor.go

Reviewed

··· 1 1 + package knotstream 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "database/sql" 6 6 + "errors" 7 7 + "log/slog" 8 8 + "strings" 9 9 + ) 10 10 + 11 11 + type hostsCursorStore struct { 12 12 + db *sql.DB 13 13 + logger *slog.Logger 14 14 + } 15 15 + 16 16 + func (s *hostsCursorStore) hostname(key string) string { 17 17 + return strings.TrimPrefix(key, "knot:") 18 18 + } 19 19 + 20 20 + func (s *hostsCursorStore) Set(key string, cursor string) { 21 21 + host := s.hostname(key) 22 22 + if _, err := s.db.ExecContext(context.Background(), 23 23 + `update hosts set last_tid = $1 where hostname = $2 and $1 > last_tid`, 24 24 + cursor, host, 25 25 + ); err != nil { 26 26 + s.logger.Error("hosts cursor set failed", "host", host, "cursor", cursor, "err", err) 27 27 + } 28 28 + } 29 29 + 30 30 + func (s *hostsCursorStore) Get(key string) string { 31 31 + host := s.hostname(key) 32 32 + var tid string 33 33 + err := s.db.QueryRowContext(context.Background(), 34 34 + `select last_tid from hosts where hostname = $1`, 35 35 + host, 36 36 + ).Scan(&tid) 37 37 + if err != nil { 38 38 + if !errors.Is(err, sql.ErrNoRows) { 39 39 + s.logger.Error("hosts cursor get failed", "host", host, "err", err) 40 40 + } 41 41 + return "" 42 42 + } 43 43 + return tid 44 44 + }

+2 -2

knotmirror/knotstream/knotstream.go

Reviewed

··· 29 29 } 30 30 31 31 func (s *KnotStream) Start(ctx context.Context) { 32 32 - go s.slurper.Run(ctx) 32 32 + s.slurper.Start(ctx) 33 33 } 34 34 35 35 func (s *KnotStream) Shutdown(ctx context.Context) error { ··· 53 53 Hostname: hostname, 54 54 NoSSL: noSSL, 55 55 Status: models.HostStatusActive, 56 56 - LastSeq: 0, 56 56 + LastTid: "", 57 57 } 58 58 59 59 if err := db.UpsertHost(ctx, s.db, host); err != nil {

+1 -13

knotmirror/knotstream/scheduler.go

Reviewed

··· 4 4 "context" 5 5 "log/slog" 6 6 "sync" 7 7 - "sync/atomic" 8 8 - "time" 9 7 10 8 "tangled.org/core/log" 11 9 ) ··· 18 16 feeder chan *Task 19 17 lk sync.Mutex 20 18 scheduled map[string][]*Task 21 21 - lastSeq atomic.Int64 22 19 23 20 logger *slog.Logger 24 21 } 25 22 26 23 type Task struct { 27 27 - Key string 28 28 - message []byte 24 24 + Key string 29 25 } 30 26 31 27 func NewParallelScheduler(maxC int, ident string, do func(context.Context, *Task) error) *ParallelScheduler { ··· 47 43 func (s *ParallelScheduler) AddTask(ctx context.Context, task *Task) { 48 44 s.lk.Lock() 49 45 if st, ok := s.scheduled[task.Key]; ok { 50 50 - // schedule task 51 46 s.scheduled[task.Key] = append(st, task) 52 47 s.lk.Unlock() 53 48 return ··· 88 83 task = rem[0] 89 84 s.scheduled[task.Key] = rem[1:] 90 85 } 91 91 - 92 92 - // TODO: update seq from received message 93 93 - s.lastSeq.Store(time.Now().UnixNano()) 94 86 }() 95 87 s.lk.Unlock() 96 88 } 97 89 } 98 90 } 99 99 - 100 100 - func (s *ParallelScheduler) LastSeq() int64 { 101 101 - return s.lastSeq.Load() 102 102 - }

+131 -236

knotmirror/knotstream/slurper.go

Reviewed

··· 6 6 "encoding/json" 7 7 "fmt" 8 8 "log/slog" 9 9 - "math/rand" 9 9 + "net" 10 10 "net/http" 11 11 + "net/url" 11 12 "sync" 12 13 "time" 13 14 ··· 15 16 "github.com/bluesky-social/indigo/util/ssrf" 16 17 "github.com/carlmjohnson/versioninfo" 17 18 "github.com/gorilla/websocket" 19 19 + "tangled.org/core/api/tangled" 20 20 + "tangled.org/core/eventconsumer" 21 21 + "tangled.org/core/eventstream" 18 22 "tangled.org/core/knotmirror/config" 19 23 "tangled.org/core/knotmirror/db" 20 24 "tangled.org/core/knotmirror/models" ··· 24 28 type KnotSlurper struct { 25 29 logger *slog.Logger 26 30 db *sql.DB 27 27 - cfg config.SlurperConfig 28 31 ssrf bool 29 32 30 30 - subsLk sync.Mutex 31 31 - subs map[string]*subscription 33 33 + hostsMu sync.RWMutex 34 34 + hosts map[string]models.Host 35 35 + 36 36 + consumer *eventconsumer.Consumer 37 37 + 38 38 + ctxMu sync.Mutex 39 39 + rootCtx context.Context 32 40 } 33 41 34 34 - func NewKnotSlurper(l *slog.Logger, db *sql.DB, cfg *config.Config) *KnotSlurper { 35 35 - return &KnotSlurper{ 36 36 - logger: log.SubLogger(l, "slurper"), 37 37 - db: db, 38 38 - cfg: cfg.Slurper, 42 42 + func NewKnotSlurper(l *slog.Logger, sqlDB *sql.DB, cfg *config.Config) *KnotSlurper { 43 43 + logger := log.SubLogger(l, "slurper") 44 44 + s := &KnotSlurper{ 45 45 + logger: logger, 46 46 + db: sqlDB, 39 47 ssrf: cfg.KnotSSRF, 40 40 - subs: make(map[string]*subscription), 48 48 + hosts: make(map[string]models.Host), 41 49 } 42 42 - } 43 50 44 44 - func (s *KnotSlurper) Run(ctx context.Context) { 45 45 - for { 46 46 - select { 47 47 - case <-ctx.Done(): 48 48 - return 49 49 - case <-time.After(s.cfg.PersistCursorPeriod): 50 50 - if err := s.persistCursors(ctx); err != nil { 51 51 - s.logger.Error("failed to flush cursors", "err", err) 52 52 - } 53 53 - } 51 51 + dialer := &websocket.Dialer{ 52 52 + HandshakeTimeout: 5 * time.Second, 53 53 + NetDialContext: s.dialContext, 54 54 } 55 55 - } 56 55 57 57 - func (s *KnotSlurper) CheckIfSubscribed(hostname string) bool { 58 58 - s.subsLk.Lock() 59 59 - defer s.subsLk.Unlock() 56 56 + store := &hostsCursorStore{db: sqlDB, logger: logger} 60 57 61 61 - _, ok := s.subs[hostname] 62 62 - return ok 58 58 + headers := http.Header{} 59 59 + headers.Set("User-Agent", userAgent()) 60 60 + 61 61 + s.consumer = eventconsumer.NewConsumer(eventconsumer.ConsumerConfig{ 62 62 + Logger: logger, 63 63 + ProcessFunc: s.processEvent, 64 64 + Dialer: dialer, 65 65 + URLFunc: s.urlFor, 66 66 + RequestHeader: headers, 67 67 + CursorStore: store, 68 68 + MaxRetryAttempts: 30, 69 69 + OnConnectExceeded: s.banHost, 70 70 + RetryInterval: time.Second, 71 71 + MaxRetryInterval: 30 * time.Second, 72 72 + ConnectionTimeout: 5 * time.Second, 73 73 + WorkerCount: cfg.Slurper.ConcurrencyPerHost, 74 74 + QueueSize: 1024, 75 75 + }) 76 76 + return s 77 77 + } 78 78 + 79 79 + func (s *KnotSlurper) Start(ctx context.Context) { 80 80 + s.ctxMu.Lock() 81 81 + s.rootCtx = ctx 82 82 + s.ctxMu.Unlock() 83 83 + s.consumer.Start(ctx) 63 84 } 64 85 65 86 func (s *KnotSlurper) Shutdown(ctx context.Context) error { 66 66 - s.logger.Info("starting shutdown host cursor flush") 67 67 - err := s.persistCursors(ctx) 68 68 - if err != nil { 69 69 - s.logger.Error("shutdown error", "err", err) 70 70 - } 87 87 + s.consumer.Stop() 71 88 s.logger.Info("slurper shutdown complete") 72 72 - return err 89 89 + return nil 73 90 } 74 91 75 75 - func (s *KnotSlurper) persistCursors(ctx context.Context) error { 76 76 - // // gather cursor list from subscriptions and store them to DB 77 77 - // start := time.Now() 78 78 - 79 79 - s.subsLk.Lock() 80 80 - cursors := make([]models.HostCursor, len(s.subs)) 81 81 - i := 0 82 82 - for _, sub := range s.subs { 83 83 - cursors[i] = sub.HostCursor() 84 84 - i++ 85 85 - } 86 86 - s.subsLk.Unlock() 87 87 - 88 88 - err := db.StoreCursors(ctx, s.db, cursors) 89 89 - // s.logger.Info("finished persisting cursors", "count", len(cursors), "duration", time.Since(start).String(), "err", err) 90 90 - return err 92 92 + func (s *KnotSlurper) CheckIfSubscribed(hostname string) bool { 93 93 + s.hostsMu.RLock() 94 94 + defer s.hostsMu.RUnlock() 95 95 + _, ok := s.hosts[hostname] 96 96 + return ok 91 97 } 92 98 93 99 func (s *KnotSlurper) Subscribe(host models.Host) error { 94 94 - s.subsLk.Lock() 95 95 - defer s.subsLk.Unlock() 96 96 - 97 97 - _, ok := s.subs[host.Hostname] 98 98 - if ok { 100 100 + s.hostsMu.Lock() 101 101 + if _, ok := s.hosts[host.Hostname]; ok { 102 102 + s.hostsMu.Unlock() 99 103 return fmt.Errorf("already subscribed: %s", host.Hostname) 100 104 } 101 101 - 102 102 - // TODO: include `cancel` function to kill subscription by hostname 103 103 - sub := &subscription{ 104 104 - hostname: host.Hostname, 105 105 - scheduler: NewParallelScheduler( 106 106 - s.cfg.ConcurrencyPerHost, 107 107 - host.Hostname, 108 108 - s.ProcessEvent, 109 109 - ), 110 110 - } 111 111 - s.subs[host.Hostname] = sub 105 105 + s.hosts[host.Hostname] = host 106 106 + s.hostsMu.Unlock() 112 107 113 113 - // TODO: use service level context, not the top-most one. 114 114 - // Using top-most context should be avoided to do graceful shutdown. 115 115 - ctx := context.TODO() 116 116 - 117 117 - sub.scheduler.Start(ctx) 118 118 - go s.subscribeWithRedialer(ctx, host, sub) 108 108 + connectedInbound.Inc() 109 109 + s.consumer.AddSource(s.subscribeCtx(), eventconsumer.NewKnotSource(host.Hostname)) 119 110 return nil 120 111 } 121 112 122 122 - func (s *KnotSlurper) subscribeWithRedialer(ctx context.Context, host models.Host, sub *subscription) { 123 123 - l := s.logger.With("host", host.Hostname) 124 124 - defer func() { 125 125 - s.subsLk.Lock() 126 126 - defer s.subsLk.Unlock() 113 113 + func (s *KnotSlurper) subscribeCtx() context.Context { 114 114 + s.ctxMu.Lock() 115 115 + defer s.ctxMu.Unlock() 116 116 + if s.rootCtx != nil { 117 117 + return s.rootCtx 118 118 + } 119 119 + return context.Background() 120 120 + } 127 121 128 128 - l.Info("unsubscribing knot") 129 129 - delete(s.subs, host.Hostname) 130 130 - }() 122 122 + func (s *KnotSlurper) banHost(src eventconsumer.Source, err error) { 123 123 + s.logger.Warn("host does not appear to be online, disabling", "host", src.Host, "err", err) 124 124 + connectedInbound.Dec() 131 125 132 132 - dialer := websocket.Dialer{ 133 133 - HandshakeTimeout: time.Second * 5, 126 126 + s.hostsMu.Lock() 127 127 + h, ok := s.hosts[src.Host] 128 128 + if ok { 129 129 + delete(s.hosts, src.Host) 134 130 } 135 135 - 136 136 - // if this isn't a localhost / private connection, then we should enable SSRF protections 137 137 - if !host.NoSSL || s.ssrf { 138 138 - netDialer := ssrf.PublicOnlyDialer() 139 139 - dialer.NetDialContext = netDialer.DialContext 131 131 + s.hostsMu.Unlock() 132 132 + if !ok { 133 133 + return 140 134 } 135 135 + h.Status = models.HostStatusOffline 136 136 + if err := db.UpsertHost(context.Background(), s.db, &h); err != nil { 137 137 + s.logger.Error("failed to update host status", "err", err) 138 138 + } 139 139 + s.consumer.RemoveSource(src) 140 140 + } 141 141 142 142 - cursor := host.LastSeq 143 143 - 144 144 - connectedInbound.Inc() 145 145 - defer connectedInbound.Dec() 146 146 - 147 147 - var backoff int 148 148 - for { 149 149 - select { 150 150 - case <-ctx.Done(): 151 151 - return 152 152 - default: 153 153 - } 154 154 - u := host.LegacyEventsURL(cursor) 155 155 - l.Debug("made url with cursor", "cursor", cursor, "url", u) 156 156 - 157 157 - // NOTE: manual backoff retry implementation to explicitly handle fails 158 158 - hdr := make(http.Header) 159 159 - hdr.Add("User-Agent", userAgent()) 160 160 - conn, resp, err := dialer.DialContext(ctx, u, hdr) 161 161 - if err != nil { 162 162 - l.Warn("dialing failed", "err", err, "backoff", backoff) 163 163 - time.Sleep(sleepForBackoff(backoff)) 164 164 - backoff++ 165 165 - if backoff > 30 { 166 166 - l.Warn("host does not appear to be online, disabling for now") 167 167 - host.Status = models.HostStatusOffline 168 168 - if err := db.UpsertHost(ctx, s.db, &host); err != nil { 169 169 - l.Error("failed to update host status", "err", err) 170 170 - } 171 171 - return 172 172 - } 173 173 - continue 174 174 - } 175 175 - 176 176 - l.Debug("knot event subscription response", "code", resp.StatusCode, "url", u) 177 177 - 178 178 - if err := s.handleConnection(ctx, conn, sub); err != nil { 179 179 - // 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 180 180 - l.Warn("host connection failed", "err", err, "backoff", backoff) 181 181 - } 182 182 - 183 183 - updatedCursor := sub.LastSeq() 184 184 - didProgress := updatedCursor > cursor 185 185 - l.Debug("cursor compare", "cursor", cursor, "updatedCursor", updatedCursor, "didProgress", didProgress) 186 186 - if cursor == 0 || didProgress { 187 187 - cursor = updatedCursor 188 188 - backoff = 0 189 189 - 190 190 - batch := []models.HostCursor{sub.HostCursor()} 191 191 - if err := db.StoreCursors(ctx, s.db, batch); err != nil { 192 192 - l.Error("failed to store cursors", "err", err) 193 193 - } 194 194 - } 142 142 + func (s *KnotSlurper) dialContext(ctx context.Context, network, addr string) (net.Conn, error) { 143 143 + host, _, err := net.SplitHostPort(addr) 144 144 + if err != nil { 145 145 + host = addr 146 146 + } 147 147 + s.hostsMu.RLock() 148 148 + h, ok := s.hosts[host] 149 149 + s.hostsMu.RUnlock() 150 150 + useSSRF := s.ssrf || !ok || !h.NoSSL 151 151 + if useSSRF { 152 152 + return ssrf.PublicOnlyDialer().DialContext(ctx, network, addr) 195 153 } 154 154 + return (&net.Dialer{}).DialContext(ctx, network, addr) 196 155 } 197 156 198 198 - // handleConnection handles websocket connection. 199 199 - // Schedules task from received event and return when connection is closed 200 200 - func (s *KnotSlurper) handleConnection(ctx context.Context, conn *websocket.Conn, sub *subscription) error { 201 201 - // ping on every 30s 202 202 - ctx, cancel := context.WithCancel(ctx) 203 203 - defer cancel() // close the background ping job on connection close 204 204 - go func() { 205 205 - t := time.NewTicker(30 * time.Second) 206 206 - defer t.Stop() 207 207 - failcount := 0 208 208 - 209 209 - for { 210 210 - select { 211 211 - case <-t.C: 212 212 - if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second*10)); err != nil { 213 213 - s.logger.Warn("failed to ping", "err", err) 214 214 - failcount++ 215 215 - if failcount >= 4 { 216 216 - s.logger.Error("too many ping fails", "count", failcount) 217 217 - _ = conn.Close() 218 218 - return 219 219 - } 220 220 - } else { 221 221 - failcount = 0 // ok ping 222 222 - } 223 223 - case <-ctx.Done(): 224 224 - _ = conn.Close() 225 225 - return 226 226 - } 227 227 - } 228 228 - }() 229 229 - 230 230 - conn.SetPingHandler(func(message string) error { 231 231 - err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Minute)) 232 232 - if err == websocket.ErrCloseSent { 233 233 - return nil 234 234 - } 235 235 - return err 236 236 - }) 237 237 - conn.SetPongHandler(func(_ string) error { 238 238 - if err := conn.SetReadDeadline(time.Now().Add(time.Minute)); err != nil { 239 239 - s.logger.Error("failed to set read deadline", "err", err) 240 240 - } 241 241 - return nil 242 242 - }) 243 243 - 244 244 - for { 245 245 - select { 246 246 - case <-ctx.Done(): 247 247 - return ctx.Err() 248 248 - default: 249 249 - } 250 250 - msgType, msg, err := conn.ReadMessage() 251 251 - if err != nil { 252 252 - return err 253 253 - } 254 254 - 255 255 - if msgType != websocket.TextMessage { 256 256 - continue 257 257 - } 258 258 - 259 259 - sub.scheduler.AddTask(ctx, &Task{ 260 260 - Key: sub.hostname, // TODO: replace to repository AT-URI for better concurrency 261 261 - message: msg, 262 262 - }) 157 157 + func (s *KnotSlurper) urlFor(src eventconsumer.Source, cursor string) (*url.URL, error) { 158 158 + s.hostsMu.RLock() 159 159 + h, ok := s.hosts[src.Host] 160 160 + s.hostsMu.RUnlock() 161 161 + scheme := "wss" 162 162 + if ok && h.NoSSL { 163 163 + scheme = "ws" 164 164 + } 165 165 + u, err := url.Parse(scheme + "://" + src.Host + "/events") 166 166 + if err != nil { 167 167 + return nil, err 168 168 + } 169 169 + if cursor != "" { 170 170 + q := url.Values{} 171 171 + q.Add("cursor", cursor) 172 172 + u.RawQuery = q.Encode() 263 173 } 174 174 + return u, nil 264 175 } 265 176 266 177 type legacyGitRefUpdate struct { ··· 269 180 LegacyRepoDid *string `json:"repoDid,omitempty"` 270 181 } 271 182 272 272 - type LegacyGitEvent struct { 273 273 - Rkey string 274 274 - Nsid string 275 275 - Event legacyGitRefUpdate 276 276 - } 277 277 - 278 278 - func (s *KnotSlurper) ProcessEvent(ctx context.Context, task *Task) error { 279 279 - var legacyMessage LegacyGitEvent 280 280 - if err := json.Unmarshal(task.message, &legacyMessage); err != nil { 281 281 - return fmt.Errorf("unmarshaling message: %w", err) 183 183 + func (s *KnotSlurper) processEvent(ctx context.Context, src eventconsumer.Source, ev eventstream.Event) error { 184 184 + if ev.Nsid != tangled.GitRefUpdateNSID { 185 185 + knotstreamEventsSkipped.Inc() 186 186 + return nil 282 187 } 283 283 - 284 284 - if err := s.ProcessLegacyGitRefUpdate(ctx, task.Key, &legacyMessage); err != nil { 285 285 - return fmt.Errorf("processing gitRefUpdate: %w", err) 286 286 - } 287 287 - return nil 188 188 + return s.processLegacyGitRefUpdate(ctx, src.Host, &ev) 288 189 } 289 190 290 191 // lookupRepoForRefUpdate resolves the local repo row for an incoming refUpdate 291 192 // via the stable RepoDid join. Returns (nil, "", nil) when the event has no 292 193 // repoDid (unjoinable) and (nil, key, nil) on a clean miss. 293 293 - func (s *KnotSlurper) lookupRepoForRefUpdate(ctx context.Context, evt *LegacyGitEvent) (*models.Repo, string, error) { 294 294 - raw := evt.Event.Repo 194 194 + func (s *KnotSlurper) lookupRepoForRefUpdate(ctx context.Context, payload legacyGitRefUpdate) (*models.Repo, string, error) { 195 195 + raw := payload.Repo 295 196 if raw == nil || *raw == "" { 296 296 - raw = evt.Event.LegacyRepoDid 197 197 + raw = payload.LegacyRepoDid 297 198 } 298 199 if raw == nil || *raw == "" { 299 200 return nil, "", nil ··· 303 204 return curr, repoDid.String(), err 304 205 } 305 206 306 306 - func (s *KnotSlurper) ProcessLegacyGitRefUpdate(ctx context.Context, source string, evt *LegacyGitEvent) error { 207 207 + func (s *KnotSlurper) processLegacyGitRefUpdate(ctx context.Context, source string, evt *eventstream.Event) error { 307 208 knotstreamEventsReceived.Inc() 308 209 309 210 l := s.logger.With("src", source) 310 211 311 311 - curr, lookupKey, err := s.lookupRepoForRefUpdate(ctx, evt) 212 212 + var payload legacyGitRefUpdate 213 213 + if err := json.Unmarshal(evt.EventJson, &payload); err != nil { 214 214 + return fmt.Errorf("decoding gitRefUpdate payload: %w", err) 215 215 + } 216 216 + 217 217 + curr, lookupKey, err := s.lookupRepoForRefUpdate(ctx, payload) 312 218 if err != nil { 313 219 return fmt.Errorf("failed to get repo '%s': %w", lookupKey, err) 314 220 } 315 221 if curr == nil { 316 222 if lookupKey == "" { 317 223 l.Warn("skipping gitRefUpdate: event has no fields to join on", 318 318 - "repo", evt.Event.Repo, "legacy_repo_did", evt.Event.LegacyRepoDid) 224 224 + "repo", payload.Repo, "legacy_repo_did", payload.LegacyRepoDid) 319 225 } else { 320 226 // if repo doesn't exist in DB, just ignore the event. That repo is unknown. 321 227 // Hopefully crawler/tap will sync it later. ··· 339 245 return nil 340 246 } 341 247 342 342 - // can't skip anything, update repo state 343 248 if err := db.UpdateRepoState(ctx, s.db, curr.RepoDid, models.RepoStateDesynchronized); err != nil { 344 249 return err 345 250 } ··· 353 258 func userAgent() string { 354 259 return fmt.Sprintf("knotmirror/%s", versioninfo.Short()) 355 260 } 356 356 - 357 357 - func sleepForBackoff(b int) time.Duration { 358 358 - if b == 0 { 359 359 - return 0 360 360 - } 361 361 - if b < 10 { 362 362 - return time.Millisecond * time.Duration((50*b)+rand.Intn(500)) 363 363 - } 364 364 - return time.Second * 30 365 365 - }

-22

knotmirror/knotstream/subscription.go

Reviewed

··· 1 1 - package knotstream 2 2 - 3 3 - import "tangled.org/core/knotmirror/models" 4 4 - 5 5 - // subscription represents websocket connection with that host 6 6 - type subscription struct { 7 7 - hostname string 8 8 - 9 9 - // embedded parallel job scheduler 10 10 - scheduler *ParallelScheduler 11 11 - } 12 12 - 13 13 - func (s *subscription) LastSeq() int64 { 14 14 - return s.scheduler.LastSeq() 15 15 - } 16 16 - 17 17 - func (s *subscription) HostCursor() models.HostCursor { 18 18 - return models.HostCursor{ 19 19 - Hostname: s.hostname, 20 20 - LastSeq: s.LastSeq(), 21 21 - } 22 22 - }

+4 -21

knotmirror/models/models.go

Reviewed

··· 56 56 return s == RepoStateResyncing 57 57 } 58 58 59 59 - type HostCursor struct { 60 60 - Hostname string 61 61 - LastSeq int64 62 62 - } 63 63 - 64 59 type Host struct { 65 60 Hostname string 66 61 NoSSL bool 67 62 Status HostStatus 68 68 - LastSeq int64 63 63 + LastTid string 69 64 } 70 65 71 66 type HostStatus string ··· 102 97 } 103 98 } 104 99 105 105 - // func (h *Host) SubscribeGitRefsURL(cursor int64) string { 106 106 - // scheme := "wss" 107 107 - // if h.NoSSL { 108 108 - // scheme = "ws" 109 109 - // } 110 110 - // u := fmt.Sprintf("%s://%s/xrpc/%s", scheme, h.Hostname, tangled.SubscribeGitRefsNSID) 111 111 - // if cursor > 0 { 112 112 - // u = fmt.Sprintf("%s?cursor=%d", u, h.LastSeq) 113 113 - // } 114 114 - // return u 115 115 - // } 116 116 - 117 117 - func (h *Host) LegacyEventsURL(cursor int64) string { 100 100 + func (h *Host) LegacyEventsURL(cursor string) string { 118 101 u := fmt.Sprintf("%s/events", h.WsURL()) 119 119 - if cursor > 0 { 120 120 - u = fmt.Sprintf("%s?cursor=%d", u, cursor) 102 102 + if cursor != "" { 103 103 + u = fmt.Sprintf("%s?cursor=%s", u, cursor) 121 104 } 122 105 return u 123 106 }