+3 -2

knotmirror/config/config.go

··· 10 10 type Config struct { 11 11 TapUrl string `env:"MIRROR_TAP_URL, default=http://localhost:2480"` 12 12 DbPath string `env:"MIRROR_DB_PATH, default=mirror.db"` 13 13 - KnotUseSSL bool `env:"MIRROR_KNOT_USE_SSL, default=false"` // use SSL for Knot when not schema is not specified 13 13 + KnotUseSSL bool `env:"MIRROR_KNOT_USE_SSL, default=false"` // use SSL for Knot when not scheme is not specified 14 14 + KnotSSRF bool `env:"MIRROR_KNOT_SSRF, default=false"` 14 15 GitRepoBasePath string `env:"MIRROR_GIT_BASEPATH, default=repos"` 15 16 GitRepoFetchTimeout time.Duration `env:"MIRROR_GIT_FETCH_TIMEOUT, default=600s"` 16 17 ResyncParallelism int `env:"MIRROR_RESYNC_PARALLELISM, default=5"` ··· 21 22 22 23 type SlurperConfig struct { 23 24 PersistCursorPeriod time.Duration `env:"PERSIST_CURSOR_PERIOD, default=4s"` 24 24 - ConcurrencyPerHost int `env:"CONCURRENCY, default=40"` 25 25 + ConcurrencyPerHost int `env:"CONCURRENCY, default=4"` 25 26 } 26 27 27 28 func Load(ctx context.Context) (*Config, error) {

+3

knotmirror/db/repos.go

··· 111 111 &repo.RetryCount, 112 112 &repo.RetryAfter, 113 113 ); err != nil { 114 114 + if errors.Is(err, sql.ErrNoRows) { 115 115 + return nil, nil 116 116 + } 114 117 return nil, fmt.Errorf("querying repo: %w", err) 115 118 } 116 119 return &repo, nil

+68

knotmirror/git.go

··· 1 1 + package knotmirror 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "fmt" 7 7 + "os/exec" 8 8 + 9 9 + "github.com/go-git/go-git/v5" 10 10 + gitconfig "github.com/go-git/go-git/v5/config" 11 11 + "github.com/go-git/go-git/v5/plumbing/transport" 12 12 + ) 13 13 + 14 14 + type GitMirrorClient interface { 15 15 + Clone(ctx context.Context, path, url string) error 16 16 + Fetch(ctx context.Context, path, url string) error 17 17 + } 18 18 + 19 19 + type CliGitMirrorClient struct{} 20 20 + 21 21 + var _ GitMirrorClient = new(CliGitMirrorClient) 22 22 + 23 23 + func (c *CliGitMirrorClient) Clone(ctx context.Context, path, url string) error { 24 24 + cmd := exec.Command("git", "clone", "--mirror", url, path) 25 25 + if out, err := cmd.CombinedOutput(); err != nil { 26 26 + return fmt.Errorf("cloning repo: %w\n%s", err, string(out)) 27 27 + } 28 28 + return nil 29 29 + } 30 30 + 31 31 + func (c *CliGitMirrorClient) Fetch(ctx context.Context, path, url string) error { 32 32 + cmd := exec.Command("git", "-C", path, "fetch", "--prune", url) 33 33 + if out, err := cmd.CombinedOutput(); err != nil { 34 34 + return fmt.Errorf("fetching repo: %w\n%s", err, string(out)) 35 35 + } 36 36 + return nil 37 37 + } 38 38 + 39 39 + type GoGitMirrorClient struct{} 40 40 + 41 41 + var _ GitMirrorClient = new(GoGitMirrorClient) 42 42 + 43 43 + func (c *GoGitMirrorClient) Clone(ctx context.Context, path string, url string) error { 44 44 + _, err := git.PlainCloneContext(ctx, path, true, &git.CloneOptions{ 45 45 + URL: url, 46 46 + Mirror: true, 47 47 + }) 48 48 + if err != nil && !errors.Is(err, transport.ErrEmptyRemoteRepository) { 49 49 + return fmt.Errorf("cloning repo: %w", err) 50 50 + } 51 51 + return nil 52 52 + } 53 53 + 54 54 + func (c *GoGitMirrorClient) Fetch(ctx context.Context, path string, url string) error { 55 55 + gr, err := git.PlainOpen(path) 56 56 + if err != nil { 57 57 + return fmt.Errorf("opening local repo: %w", err) 58 58 + } 59 59 + if err := gr.FetchContext(ctx, &git.FetchOptions{ 60 60 + RemoteURL: url, 61 61 + RefSpecs: []gitconfig.RefSpec{gitconfig.RefSpec("+refs/*:refs/*")}, 62 62 + Force: true, 63 63 + Prune: true, 64 64 + }); err != nil { 65 65 + return fmt.Errorf("fetching reppo: %w", err) 66 66 + } 67 67 + return nil 68 68 + }

+16

knotmirror/knotmirror.go

··· 4 4 "context" 5 5 "fmt" 6 6 "net/http" 7 7 + _ "net/http/pprof" 7 8 "time" 8 9 9 10 "github.com/prometheus/client_golang/prometheus/promhttp" 10 11 "tangled.org/core/knotmirror/config" 11 12 "tangled.org/core/knotmirror/db" 12 13 "tangled.org/core/knotmirror/knotstream" 14 14 + "tangled.org/core/knotmirror/models" 13 15 "tangled.org/core/log" 14 16 ) 15 17 ··· 30 32 if err != nil { 31 33 return fmt.Errorf("initializing db: %w", err) 32 34 } 35 35 + 36 36 + res, err := db.ExecContext(ctx, 37 37 + `update repos set state = ? where state = ?`, 38 38 + models.RepoStateDesynchronized, 39 39 + models.RepoStateResyncing, 40 40 + ) 41 41 + if err != nil { 42 42 + return fmt.Errorf("clearing resyning states: %w", err) 43 43 + } 44 44 + rows, err := res.RowsAffected() 45 45 + if err != nil { 46 46 + return fmt.Errorf("getting affected rows: %w", err) 47 47 + } 48 48 + logger.Info(fmt.Sprintf("clearing resyning states: %d records updated", rows)) 33 49 34 50 knotstream := knotstream.NewKnotStream(logger, db, cfg) 35 51 crawler := NewCrawler(logger, db)

+3 -2

knotmirror/knotstream/knotstream.go

··· 42 42 } 43 43 44 44 func (s *KnotStream) SubscribeHost(ctx context.Context, hostname string, noSSL bool) error { 45 45 - s.logger.Debug("subscribe", "nossl", noSSL) 45 45 + l := s.logger.With("hostname", hostname, "nossl", noSSL) 46 46 + l.Debug("subscribe") 46 47 host, err := db.GetHost(ctx, s.db, hostname) 47 48 if err != nil { 48 49 return fmt.Errorf("loading host from db: %w", err) ··· 60 61 return fmt.Errorf("adding host to db: %w", err) 61 62 } 62 63 63 63 - s.logger.Info("adding new host subscription", "hostname", hostname, "noSSL", noSSL) 64 64 + l.Info("adding new host subscription") 64 65 } 65 66 66 67 if host.Status == models.HostStatusBanned {

+2 -2

knotmirror/readme.md

··· 6 6 7 7 # TODO 8 8 9 9 - - [ ] cleanup 'resyncing' state on shutdown (or on startup too) 10 10 - - [ ] better tap reconnecting logic 9 9 + - [x] cleanup 'resyncing' state on shutdown (or on startup too) 10 10 + - [x] better tap reconnecting logic 11 11 - [ ] handle really large repos (maybe shallow-clone first?) 12 12 13 13 idea: run multiple different resync workers. 4 for long running tasks, 10 for short tasks. on timeout, schedule it for long running task

+43 -37

knotmirror/resyncer.go

··· 8 8 "log/slog" 9 9 "math/rand" 10 10 "net/url" 11 11 + "os" 11 12 "path" 12 13 "sync" 13 14 "time" 14 15 15 16 "github.com/bluesky-social/indigo/atproto/syntax" 16 16 - "github.com/go-git/go-git/v5" 17 17 - gitconfig "github.com/go-git/go-git/v5/config" 18 18 - "github.com/go-git/go-git/v5/plumbing/transport" 19 17 "tangled.org/core/knotmirror/config" 20 18 "tangled.org/core/knotmirror/db" 21 19 "tangled.org/core/knotmirror/models" ··· 74 72 l.Info("processing resync", "aturi", repoAt) 75 73 if err := r.resyncRepo(ctx, repoAt); err != nil { 76 74 l.Error("resync failed", "aturi", repoAt, "error", err) 75 75 + 76 76 + // TODO: cleanup. set resync state to desynchronized 77 77 } 78 78 } 79 79 } ··· 143 143 } 144 144 145 145 repoPath := r.repoPath(repo) 146 146 - remoteUrl := r.repoRemoteURL(repo) 147 147 - l := r.logger.With("repo", repo.DidSlashRepo(), "path", repoPath, "url", remoteUrl) 146 146 + l := r.logger.With("repo", repo.DidSlashRepo(), "path", repoPath) 147 147 + 148 148 + remoteUrl, err := r.repoRemoteURL(repo) 149 149 + if err != nil { 150 150 + return false, fmt.Errorf("parsing knot url: %w", err) 151 151 + } 152 152 + l = l.With("url", remoteUrl) 148 153 149 154 ctx, cancel := context.WithTimeout(ctx, r.repoFetchTimeout) 150 155 defer cancel() ··· 152 157 // TODO: check if Knot is on backoff list. If so, return (false, nil) 153 158 // TODO: use r.repoFetchTimeout on fetch 154 159 // TODO: detect rate limit error (http.StatusTooManyRequests) to put Knot in backoff list 155 155 - gr, err := git.PlainOpen(repoPath) 156 156 - if errors.Is(err, git.ErrRepositoryNotExists) { 157 157 - l.Debug("cloning repo") 158 158 - // if err := exec.Command("git", "clone", "--mirror", remoteUrl, repoPath).Run(); err != nil { 159 159 - // return false, fmt.Errorf("cloning repo: %w", err) 160 160 - // } 161 161 - _, err := git.PlainCloneContext(ctx, repoPath, true, &git.CloneOptions{ 162 162 - URL: remoteUrl, 163 163 - Mirror: true, 164 164 - }) 165 165 - if err != nil && !errors.Is(err, transport.ErrEmptyRemoteRepository) { 166 166 - return false, fmt.Errorf("cloning repo: %w", err) 160 160 + 161 161 + // NOTE: using plain git-cli for clone/fetch as go-git is too memory-intensive. 162 162 + gitclient := &CliGitMirrorClient{} 163 163 + 164 164 + exist, err := isDir(repoPath) 165 165 + if err != nil { 166 166 + return false, fmt.Errorf("checking repo path: %w", err) 167 167 + } 168 168 + if !exist { 169 169 + if err := gitclient.Clone(ctx, repoPath, remoteUrl); err != nil { 170 170 + return false, err 167 171 } 168 172 } else { 169 169 - if err != nil { 170 170 - return false, fmt.Errorf("laoding repo: %w", err) 171 171 - } 172 172 - l.Debug("fetching repo") 173 173 - // if err := exec.Command("git", "-C", repoPath, "fetch", "--mirror", remoteUrl).Run(); err != nil { 174 174 - // return false, fmt.Errorf("fetching repo: %w", err) 175 175 - // } 176 176 - if err := gr.FetchContext(ctx, &git.FetchOptions{ 177 177 - RemoteURL: remoteUrl, 178 178 - RefSpecs: []gitconfig.RefSpec{gitconfig.RefSpec("+refs/*:refs/*")}, 179 179 - Force: true, 180 180 - Prune: true, 181 181 - }); err != nil { 182 182 - return false, fmt.Errorf("fetching reppo: %w", err) 173 173 + if err := gitclient.Fetch(ctx, repoPath, remoteUrl); err != nil { 174 174 + return false, err 183 175 } 184 176 } 185 177 ··· 209 201 210 202 repo, err := db.GetRepoByAtUri(ctx, r.db, repoAt) 211 203 if err != nil { 212 212 - return err 204 204 + return fmt.Errorf("failed to get repo: %w", err) 213 205 } 214 206 if repo == nil { 215 207 return fmt.Errorf("failed to get repo. repo '%s' doesn't exist in db", repoAt) ··· 217 209 218 210 var retryCount = repo.RetryCount + 1 219 211 var retryAfter int64 220 220 - if retryCount >= 5 { 212 212 + if retryCount >= 10 { 221 213 state = models.RepoStateSuspended 222 222 - errMsg = "too many resync fails" 214 214 + errMsg = fmt.Sprintf("too many resync fails: %s", errMsg) 223 215 retryAfter = 0 224 216 } else { 225 217 // start a 1 min & go up to 1 hr between retries ··· 240 232 return path.Join(r.repoBasePath, repo.Did.String(), repo.Rkey.String()) 241 233 } 242 234 243 243 - func (r *Resyncer) repoRemoteURL(repo *models.Repo) string { 244 244 - u, _ := url.Parse(repo.KnotDomain) 235 235 + func (r *Resyncer) repoRemoteURL(repo *models.Repo) (string, error) { 236 236 + u, err := url.Parse(repo.KnotDomain) 237 237 + if err != nil { 238 238 + return "", err 239 239 + } 245 240 if u.Scheme == "" { 246 241 if r.knotUseSSL { 247 242 u.Scheme = "https" ··· 250 245 } 251 246 } 252 247 u = u.JoinPath(repo.DidSlashRepo()) 253 253 - return u.String() 248 248 + return u.String(), nil 254 249 } 255 250 256 251 func backoff(retries int, max int) time.Duration { ··· 258 253 jitter := time.Millisecond * time.Duration(rand.Intn(1000)) 259 254 return time.Second*time.Duration(dur) + jitter 260 255 } 256 256 + 257 257 + func isDir(path string) (bool, error) { 258 258 + info, err := os.Stat(path) 259 259 + if err == nil && info.IsDir() { 260 260 + return true, nil 261 261 + } 262 262 + if os.IsNotExist(err) { 263 263 + return false, nil 264 264 + } 265 265 + return false, err 266 266 + }

+22 -2

knotmirror/tapclient.go

··· 6 6 "encoding/json" 7 7 "fmt" 8 8 "log/slog" 9 9 + "net/netip" 9 10 "net/url" 10 11 "time" 11 12 ··· 82 83 status = models.RepoStateSuspended 83 84 errMsg = "failed to parse knot url" 84 85 } 85 85 - if u.Hostname() == "localhost" { 86 86 + if t.cfg.KnotSSRF && isPrivate(u.Hostname()) { 86 87 status = models.RepoStateSuspended 87 87 - errMsg = "suspending localhost knot" 88 88 + errMsg = "suspending non-public knot" 88 89 } 89 90 90 91 if err := db.UpsertRepo(ctx, t.db, &models.Repo{ ··· 112 113 } 113 114 return nil 114 115 } 116 116 + 117 117 + // isPrivate checks if host is private network. It doesn't perform DNS resolution 118 118 + func isPrivate(host string) bool { 119 119 + if host == "localhost" { 120 120 + return true 121 121 + } 122 122 + addr, err := netip.ParseAddr(host) 123 123 + if err != nil { 124 124 + return false 125 125 + } 126 126 + return isPrivateAddr(addr) 127 127 + } 128 128 + 129 129 + func isPrivateAddr(addr netip.Addr) bool { 130 130 + return addr.IsLoopback() || 131 131 + addr.IsPrivate() || 132 132 + addr.IsLinkLocalUnicast() || 133 133 + addr.IsLinkLocalMulticast() 134 134 + }