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

+2 -1

flake.nix

··· 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 - knotmirror = self.callPackage ./nix/pkgs/knot-mirror.nix {}; 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 + knotmirror 136 137 ; 137 138 138 139 pkgsStatic-appview = staticPackages.appview;

+6 -4

knotmirror/knotstream/slurper.go

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

+17 -5

knotmirror/models/models.go

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

+3 -1

knotmirror/resyncer.go

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

+17 -2

knotmirror/tapclient.go

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

+14 -2

nix/modules/knotmirror.nix

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

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

+43 -1

nix/vm.nix

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

Configure Feed

Configure Feed