nix,knotmirror: listen & sync local knots from local knotmirror #1200

+2 -1

flake.nix

reviewed

··· 107 107 knot = self.callPackage ./nix/pkgs/knot.nix {}; 108 108 dolly = self.callPackage ./nix/pkgs/dolly.nix {}; 109 109 tap = self.callPackage ./nix/pkgs/tap.nix {}; 110 110 - knotmirror = self.callPackage ./nix/pkgs/knot-mirror.nix {}; 110 110 + knotmirror = self.callPackage ./nix/pkgs/knotmirror.nix {}; 111 111 }); 112 112 in { 113 113 overlays.default = final: prev: { ··· 133 133 docs 134 134 dolly 135 135 tap 136 136 + knotmirror 136 137 ; 137 138 138 139 pkgsStatic-appview = staticPackages.appview;

+6 -4

knotmirror/knotstream/slurper.go

reviewed

··· 263 263 return fmt.Errorf("unmarshaling message: %w", err) 264 264 } 265 265 266 266 - if err := s.ProcessLegacyGitRefUpdate(ctx, &legacyMessage); err != nil { 266 266 + if err := s.ProcessLegacyGitRefUpdate(ctx, task.key, &legacyMessage); err != nil { 267 267 return fmt.Errorf("processing gitRefUpdate: %w", err) 268 268 } 269 269 return nil 270 270 } 271 271 272 272 - func (s *KnotSlurper) ProcessLegacyGitRefUpdate(ctx context.Context, evt *LegacyGitEvent) error { 272 272 + func (s *KnotSlurper) ProcessLegacyGitRefUpdate(ctx context.Context, source string, evt *LegacyGitEvent) error { 273 273 knotstreamEventsReceived.Inc() 274 274 275 275 + l := s.logger.With("src", source) 276 276 + 275 277 curr, err := db.GetRepoByName(ctx, s.db, syntax.DID(evt.Event.RepoDid), evt.Event.RepoName) 276 278 if err != nil { 277 279 return fmt.Errorf("failed to get repo '%s': %w", evt.Event.RepoDid+"/"+evt.Event.RepoName, err) ··· 284 286 // But we want to store that in did/rkey in knot-mirror. 285 287 // Therefore, we should ignore when the repository is unknown. 286 288 // Hopefully crawler will sync it later. 287 287 - s.logger.Warn("skipping event from unknown repo", "did/repo", evt.Event.RepoDid+"/"+evt.Event.RepoName) 289 289 + l.Warn("skipping event from unknown repo", "did/name", evt.Event.RepoDid+"/"+evt.Event.RepoName) 288 290 knotstreamEventsSkipped.Inc() 289 291 return nil 290 292 } 291 291 - l := s.logger.With("repoAt", curr.AtUri()) 293 293 + l = l.With("repoAt", curr.AtUri()) 292 294 293 295 // TODO: should plan resync to resyncBuffer on RepoStateResyncing 294 296 if curr.State != models.RepoStateActive {

+17 -5

knotmirror/models/models.go

reviewed

··· 85 85 HostStatusBanned, 86 86 } 87 87 88 88 + func (h *Host) URL() string { 89 89 + if h.NoSSL { 90 90 + return fmt.Sprintf("http://%s", h.Hostname) 91 91 + } else { 92 92 + return fmt.Sprintf("https://%s", h.Hostname) 93 93 + } 94 94 + } 95 95 + 96 96 + func (h *Host) WsURL() string { 97 97 + if h.NoSSL { 98 98 + return fmt.Sprintf("ws://%s", h.Hostname) 99 99 + } else { 100 100 + return fmt.Sprintf("wss://%s", h.Hostname) 101 101 + } 102 102 + } 103 103 + 88 104 // func (h *Host) SubscribeGitRefsURL(cursor int64) string { 89 105 // scheme := "wss" 90 106 // if h.NoSSL { ··· 98 114 // } 99 115 100 116 func (h *Host) LegacyEventsURL(cursor int64) string { 101 101 - scheme := "wss" 102 102 - if h.NoSSL { 103 103 - scheme = "ws" 104 104 - } 105 105 - u := fmt.Sprintf("%s://%s/events", scheme, h.Hostname) 117 117 + u := fmt.Sprintf("%s/events", h.WsURL()) 106 118 if cursor > 0 { 107 119 u = fmt.Sprintf("%s?cursor=%d", u, cursor) 108 120 }

+3 -1

knotmirror/resyncer.go

reviewed

··· 24 24 logger *slog.Logger 25 25 db *sql.DB 26 26 gitm GitMirrorManager 27 27 + cfg *config.Config 27 28 28 29 claimJobMu sync.Mutex 29 30 ··· 43 44 logger: log.SubLogger(l, "resyncer"), 44 45 db: db, 45 46 gitm: gitm, 47 47 + cfg: cfg, 46 48 47 49 runningJobs: make(map[syntax.ATURI]context.CancelFunc), 48 50 ··· 272 274 273 275 // checkKnotReachability checks if Knot is reachable and is valid git remote server 274 276 func (r *Resyncer) checkKnotReachability(ctx context.Context, repo *models.Repo) error { 275 275 - repoUrl, err := makeRepoRemoteUrl(repo.KnotDomain, repo.DidSlashRepo(), true) 277 277 + repoUrl, err := makeRepoRemoteUrl(repo.KnotDomain, repo.DidSlashRepo(), r.cfg.KnotUseSSL) 276 278 if err != nil { 277 279 return err 278 280 }

+17 -2

knotmirror/tapclient.go

reviewed

··· 8 8 "log/slog" 9 9 "net/netip" 10 10 "net/url" 11 11 + "strings" 11 12 "time" 12 13 13 14 "tangled.org/core/api/tangled" ··· 78 79 return fmt.Errorf("parsing record: %w", err) 79 80 } 80 81 82 82 + knotUrl := record.Knot 83 83 + if !strings.Contains(record.Knot, "://") { 84 84 + if host, _ := db.GetHost(ctx, t.db, record.Knot); host != nil { 85 85 + knotUrl = host.URL() 86 86 + } else { 87 87 + t.logger.Warn("repo is from unknown knot") 88 88 + if t.cfg.KnotUseSSL { 89 89 + knotUrl = "https://" + knotUrl 90 90 + } else { 91 91 + knotUrl = "http://" + knotUrl 92 92 + } 93 93 + } 94 94 + } 95 95 + 81 96 status := models.RepoStatePending 82 97 errMsg := "" 83 83 - u, err := url.Parse("http://" + record.Knot) // parsing with fake scheme 98 98 + u, err := url.Parse(knotUrl) 84 99 if err != nil { 85 100 status = models.RepoStateSuspended 86 101 errMsg = "failed to parse knot url" ··· 94 109 Rkey: evt.Rkey, 95 110 Cid: evt.CID, 96 111 Name: record.Name, 97 97 - KnotDomain: record.Knot, 112 112 + KnotDomain: knotUrl, 98 113 State: status, 99 114 ErrorMsg: errMsg, 100 115 RetryAfter: 0, // clear retry info

+14 -2

nix/modules/knotmirror.nix

reviewed

··· 66 66 description = "Whether to automatically mirror from entire network"; 67 67 }; 68 68 69 69 + knotUseSSL = mkOption { 70 70 + type = types.bool; 71 71 + default = true; 72 72 + description = "Use SSL for knot connection"; 73 73 + }; 74 74 + 75 75 + knotSSRF = mkOption { 76 76 + type = types.bool; 77 77 + default = true; 78 78 + description = "enable SSRF protection for knots"; 79 79 + }; 80 80 + 69 81 tap = { 70 82 port = mkOption { 71 83 type = types.port; ··· 128 140 "MIRROR_TAP_URL=http://localhost:${toString cfg.tap.port}" 129 141 "MIRROR_DB_URL=${cfg.dbUrl}" 130 142 "MIRROR_GIT_BASEPATH=/var/lib/knotmirror/repos" 131 131 - "MIRROR_KNOT_USE_SSL=true" 132 132 - "MIRROR_KNOT_SSRF=true" 143 143 + "MIRROR_KNOT_USE_SSL=${boolToString cfg.knotUseSSL}" 144 144 + "MIRROR_KNOT_SSRF=${boolToString cfg.knotSSRF}" 133 145 "MIRROR_RESYNC_PARALLELISM=12" 134 146 "MIRROR_METRICS_LISTEN=127.0.0.1:7100" 135 147 "MIRROR_ADMIN_LISTEN=${cfg.adminListenAddr}"

nix/pkgs/knot-mirror.nix nix/pkgs/knotmirror.nix

reviewed

+43 -1

nix/vm.nix

reviewed

··· 25 25 modules = [ 26 26 self.nixosModules.knot 27 27 self.nixosModules.spindle 28 28 + self.nixosModules.knotmirror 28 29 ({ 29 30 lib, 30 31 config, ··· 57 58 host.port = 6555; 58 59 guest.port = 6555; 59 60 } 61 61 + # knotmirror 62 62 + { 63 63 + from = "host"; 64 64 + host.port = 7007; # 7000 is deserved in macos for Airplay 65 65 + guest.port = 7000; 66 66 + } 67 67 + # knotmirror-tap 68 68 + { 69 69 + from = "host"; 70 70 + host.port = 7480; 71 71 + guest.port = 7480; 72 72 + } 73 73 + # knotmirror-admin 74 74 + { 75 75 + from = "host"; 76 76 + host.port = 7200; 77 77 + guest.port = 7200; 78 78 + } 60 79 ]; 61 80 sharedDirectories = { 62 81 # We can't use the 9p mounts directly for most of these ··· 81 100 networking.firewall.enable = false; 82 101 time.timeZone = "Europe/London"; 83 102 services.getty.autologinUser = "root"; 84 84 - environment.systemPackages = with pkgs; [curl vim git sqlite litecli]; 103 103 + environment.systemPackages = with pkgs; [curl vim git sqlite litecli postgresql_14]; 85 104 services.tangled.knot = { 86 105 enable = true; 87 106 motd = "Welcome to the development knot!\n"; ··· 109 128 }; 110 129 }; 111 130 }; 131 131 + services.postgresql = { 132 132 + enable = true; 133 133 + package = pkgs.postgresql_14; 134 134 + ensureDatabases = ["mirror" "tap"]; 135 135 + ensureUsers = [ 136 136 + { name = "tnglr"; } 137 137 + ]; 138 138 + authentication = '' 139 139 + local all tnglr trust 140 140 + host all tnglr 127.0.0.1/32 trust 141 141 + ''; 142 142 + }; 143 143 + services.tangled.knotmirror = { 144 144 + enable = true; 145 145 + listenAddr = "0.0.0.0:7000"; 146 146 + adminListenAddr = "0.0.0.0:7200"; 147 147 + hostname = "localhost:7000"; 148 148 + dbUrl = "postgresql://tnglr@127.0.0.1:5432/mirror"; 149 149 + fullNetwork = false; 150 150 + tap.dbUrl = "postgresql://tnglr@127.0.0.1:5432/tap"; 151 151 + }; 112 152 users = { 113 153 # So we don't have to deal with permission clashing between 114 154 # blank disk VMs and existing state ··· 135 175 in { 136 176 knot = mkDataSyncScripts "/mnt/knot-data" config.services.tangled.knot.stateDir; 137 177 spindle = mkDataSyncScripts "/mnt/spindle-data" (builtins.dirOf config.services.tangled.spindle.server.dbPath); 178 178 + knotmirror.after = ["postgresql.target"]; 179 179 + tap-knotmirror.after = ["postgresql.target"]; 138 180 }; 139 181 }) 140 182 ];