knotserver: add repo DID migration on startup #1144

+1

appview/db/db.go

··· 34 34 "_journal_mode=WAL", 35 35 "_synchronous=NORMAL", 36 36 "_auto_vacuum=incremental", 37 37 + "_busy_timeout=5000", 37 38 } 38 39 39 40 logger := log.FromContext(ctx)

+8 -5

eventconsumer/consumer.go

··· 19 19 type ProcessFunc func(ctx context.Context, source Source, message Message) error 20 20 21 21 type Message struct { 22 22 - Rkey string 23 23 - Nsid string 24 24 - // do not full deserialize this portion of the message, processFunc can do that 22 22 + Rkey string 23 23 + Nsid string 24 24 + Created int64 `json:"created"` 25 25 EventJson json.RawMessage `json:"event"` 26 26 } 27 27 ··· 159 159 return 160 160 } 161 161 162 162 - // update cursor 163 163 - c.cfg.CursorStore.Set(j.source.Key(), time.Now().UnixNano()) 162 162 + cursorVal := msg.Created 163 163 + if cursorVal == 0 { 164 164 + cursorVal = time.Now().UnixNano() 165 165 + } 166 166 + c.cfg.CursorStore.Set(j.source.Key(), cursorVal) 164 167 165 168 if err := c.cfg.ProcessFunc(ctx, j.source, msg); err != nil { 166 169 c.logger.Error("error processing message", "source", j.source, "err", err)

+1

knotserver/db/db.go

··· 26 26 "_journal_mode=WAL", 27 27 "_synchronous=NORMAL", 28 28 "_auto_vacuum=incremental", 29 29 + "_busy_timeout=5000", 29 30 } 30 31 31 32 logger := log.FromContext(ctx)

+7 -7

knotserver/events.go

··· 44 44 } 45 45 }() 46 46 47 47 - defaultCursor := time.Now().UnixNano() 47 47 + var cursor int64 48 48 cursorStr := r.URL.Query().Get("cursor") 49 49 - cursor, err := strconv.ParseInt(cursorStr, 10, 64) 50 50 - if err != nil { 51 51 - l.Error("empty or invalid cursor", "invalidCursor", cursorStr, "default", defaultCursor) 52 52 - } 53 53 - if cursor == 0 { 54 54 - cursor = defaultCursor 49 49 + if cursorStr != "" { 50 50 + cursor, err = strconv.ParseInt(cursorStr, 10, 64) 51 51 + if err != nil { 52 52 + l.Error("invalid cursor, starting from beginning", "invalidCursor", cursorStr) 53 53 + cursor = 0 54 54 + } 55 55 } 56 56 57 57 l.Debug("going through backfill", "cursor", cursor)

+271

knotserver/migrate.go

··· 1 1 + package knotserver 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "database/sql" 6 6 + "errors" 7 7 + "fmt" 8 8 + "log/slog" 9 9 + "math/rand/v2" 10 10 + "os" 11 11 + "path/filepath" 12 12 + "strings" 13 13 + "time" 14 14 + 15 15 + "tangled.org/core/knotserver/config" 16 16 + "tangled.org/core/knotserver/db" 17 17 + "tangled.org/core/knotserver/repodid" 18 18 + "tangled.org/core/notifier" 19 19 + "tangled.org/core/rbac" 20 20 + ) 21 21 + 22 22 + type legacyRepo struct { 23 23 + ownerDid string 24 24 + repoName string 25 25 + oldPath string 26 26 + } 27 27 + 28 28 + func scanLegacyRepos(scanPath string, logger *slog.Logger) []legacyRepo { 29 29 + topEntries, err := os.ReadDir(scanPath) 30 30 + if err != nil { 31 31 + logger.Error("reading scan path", "error", err) 32 32 + return nil 33 33 + } 34 34 + 35 35 + var repos []legacyRepo 36 36 + for _, entry := range topEntries { 37 37 + if !entry.IsDir() || !strings.HasPrefix(entry.Name(), "did:") { 38 38 + continue 39 39 + } 40 40 + 41 41 + ownerPath := filepath.Join(scanPath, entry.Name()) 42 42 + 43 43 + if _, headErr := os.Stat(filepath.Join(ownerPath, "HEAD")); headErr == nil { 44 44 + continue 45 45 + } 46 46 + 47 47 + subEntries, readErr := os.ReadDir(ownerPath) 48 48 + if readErr != nil { 49 49 + logger.Error("reading owner dir", "ownerDir", ownerPath, "error", readErr) 50 50 + continue 51 51 + } 52 52 + 53 53 + ownerDid := entry.Name() 54 54 + for _, sub := range subEntries { 55 55 + if !sub.IsDir() { 56 56 + continue 57 57 + } 58 58 + subPath := filepath.Join(ownerPath, sub.Name()) 59 59 + if _, statErr := os.Stat(filepath.Join(subPath, "HEAD")); statErr != nil { 60 60 + logger.Warn("skipping non-repo directory", "path", subPath) 61 61 + continue 62 62 + } 63 63 + repos = append(repos, legacyRepo{ 64 64 + ownerDid: ownerDid, 65 65 + repoName: sub.Name(), 66 66 + oldPath: subPath, 67 67 + }) 68 68 + } 69 69 + } 70 70 + return repos 71 71 + } 72 72 + 73 73 + func migrateReposOnStartup(ctx context.Context, c *config.Config, d *db.DB, e *rbac.Enforcer, n *notifier.Notifier, logger *slog.Logger) { 74 74 + repos := scanLegacyRepos(c.Repo.ScanPath, logger) 75 75 + if len(repos) == 0 { 76 76 + logger.Info("no legacy repos found, migration complete") 77 77 + return 78 78 + } 79 79 + 80 80 + logger.Info("starting legacy repo migration", "count", len(repos)) 81 81 + start := time.Now() 82 82 + 83 83 + knotServiceUrl := "https://" + c.Server.Hostname 84 84 + if c.Server.Dev { 85 85 + knotServiceUrl = "http://" + c.Server.Hostname 86 86 + } 87 87 + 88 88 + migrated := 0 89 89 + for _, repo := range repos { 90 90 + select { 91 91 + case <-ctx.Done(): 92 92 + logger.Info("migration interrupted by shutdown", "migrated", migrated, "remaining", len(repos)-migrated) 93 93 + return 94 94 + default: 95 95 + } 96 96 + 97 97 + err := migrateOneRepo(ctx, c, d, e, n, logger, repo, knotServiceUrl) 98 98 + if err != nil { 99 99 + logger.Error("migration failed for repo", "owner", repo.ownerDid, "repo", repo.repoName, "error", err) 100 100 + continue 101 101 + } 102 102 + migrated++ 103 103 + } 104 104 + 105 105 + logger.Info("legacy repo migration complete", "migrated", migrated, "total", len(repos), "duration", time.Since(start)) 106 106 + } 107 107 + 108 108 + func migrateOneRepo( 109 109 + ctx context.Context, 110 110 + c *config.Config, 111 111 + d *db.DB, 112 112 + e *rbac.Enforcer, 113 113 + n *notifier.Notifier, 114 114 + logger *slog.Logger, 115 115 + repo legacyRepo, 116 116 + knotServiceUrl string, 117 117 + ) error { 118 118 + l := logger.With("owner", repo.ownerDid, "repo", repo.repoName) 119 119 + 120 120 + repoDid, err := d.GetRepoDid(repo.ownerDid, repo.repoName) 121 121 + needsMint := errors.Is(err, sql.ErrNoRows) 122 122 + if err != nil && !needsMint { 123 123 + return fmt.Errorf("checking repo_keys: %w", err) 124 124 + } 125 125 + 126 126 + if needsMint { 127 127 + repoDid, err = mintAndStoreRepoDID(ctx, c, d, l, repo, knotServiceUrl) 128 128 + if err != nil { 129 129 + return err 130 130 + } 131 131 + } 132 132 + 133 133 + if err := rewriteRBACPolicies(e, repo.ownerDid, repo.repoName, repoDid, l); err != nil { 134 134 + l.Error("RBAC rewrite failed (non-fatal)", "error", err) 135 135 + } 136 136 + 137 137 + newPath := filepath.Join(c.Repo.ScanPath, repoDid) 138 138 + 139 139 + if err := moveRepoOnDisk(repo.oldPath, newPath, l); err != nil { 140 140 + return err 141 141 + } 142 142 + 143 143 + ownerDir := filepath.Dir(repo.oldPath) 144 144 + if err := os.Remove(ownerDir); err != nil && !errors.Is(err, os.ErrNotExist) { 145 145 + l.Warn("could not remove empty owner dir", "path", ownerDir, "error", err) 146 146 + } 147 147 + 148 148 + if err := d.EmitDIDAssign(n, repo.ownerDid, repo.repoName, repoDid, ""); err != nil { 149 149 + l.Error("emitting didAssign event failed (non-fatal)", "error", err) 150 150 + } 151 151 + 152 152 + l.Info("migrated repo", "repoDid", repoDid) 153 153 + return nil 154 154 + } 155 155 + 156 156 + func mintAndStoreRepoDID( 157 157 + ctx context.Context, 158 158 + c *config.Config, 159 159 + d *db.DB, 160 160 + l *slog.Logger, 161 161 + repo legacyRepo, 162 162 + knotServiceUrl string, 163 163 + ) (string, error) { 164 164 + prepared, err := repodid.PrepareRepoDID(c.Server.PlcUrl, knotServiceUrl) 165 165 + if err != nil { 166 166 + return "", fmt.Errorf("preparing DID: %w", err) 167 167 + } 168 168 + 169 169 + if err := submitWithBackoff(ctx, prepared, l); err != nil { 170 170 + return "", fmt.Errorf("PLC submission: %w", err) 171 171 + } 172 172 + 173 173 + if err := d.StoreRepoKey(prepared.RepoDid, prepared.SigningKeyRaw, repo.ownerDid, repo.repoName, ""); err != nil { 174 174 + return "", fmt.Errorf("storing repo key: %w", err) 175 175 + } 176 176 + 177 177 + return prepared.RepoDid, nil 178 178 + } 179 179 + 180 180 + func submitWithBackoff(ctx context.Context, prepared *repodid.PreparedDID, l *slog.Logger) error { 181 181 + backoff := 2 * time.Second 182 182 + const maxBackoff = 5 * time.Minute 183 183 + const maxAttempts = 20 184 184 + 185 185 + for attempt := range maxAttempts { 186 186 + plcCtx, cancel := context.WithTimeout(ctx, 30*time.Second) 187 187 + err := prepared.Submit(plcCtx) 188 188 + cancel() 189 189 + 190 190 + if err == nil { 191 191 + return nil 192 192 + } 193 193 + 194 194 + errMsg := err.Error() 195 195 + retryable := strings.Contains(errMsg, "429") || 196 196 + strings.Contains(errMsg, "500") || 197 197 + strings.Contains(errMsg, "502") || 198 198 + strings.Contains(errMsg, "503") || 199 199 + strings.Contains(errMsg, "504") 200 200 + 201 201 + if !retryable || attempt == maxAttempts-1 { 202 202 + return err 203 203 + } 204 204 + 205 205 + jitter := time.Duration(rand.Int64N(int64(backoff / 4))) 206 206 + delay := backoff + jitter 207 207 + 208 208 + l.Info("PLC rate limited, backing off", "delay", delay, "attempt", attempt+1) 209 209 + 210 210 + select { 211 211 + case <-ctx.Done(): 212 212 + return ctx.Err() 213 213 + case <-time.After(delay): 214 214 + } 215 215 + 216 216 + backoff = min(backoff*2, maxBackoff) 217 217 + } 218 218 + 219 219 + return fmt.Errorf("unreachable: exceeded max attempts") 220 220 + } 221 221 + 222 222 + func moveRepoOnDisk(oldPath, newPath string, l *slog.Logger) error { 223 223 + if _, err := os.Stat(newPath); err == nil { 224 224 + l.Info("new path already exists, skipping rename", "newPath", newPath) 225 225 + return nil 226 226 + } 227 227 + 228 228 + if _, err := os.Stat(oldPath); errors.Is(err, os.ErrNotExist) { 229 229 + return fmt.Errorf("old path %s does not exist and new path %s not found", oldPath, newPath) 230 230 + } 231 231 + 232 232 + if err := os.Rename(oldPath, newPath); err != nil { 233 233 + return fmt.Errorf("renaming %s -> %s: %w", oldPath, newPath, err) 234 234 + } 235 235 + 236 236 + return nil 237 237 + } 238 238 + 239 239 + func rewriteRBACPolicies(e *rbac.Enforcer, ownerDid, repoName, repoDid string, l *slog.Logger) error { 240 240 + oldResource := ownerDid + "/" + repoName 241 241 + 242 242 + policies, err := e.E.GetFilteredPolicy(1, rbac.ThisServer, oldResource) 243 243 + if err != nil { 244 244 + return fmt.Errorf("getting old policies: %w", err) 245 245 + } 246 246 + 247 247 + var addPolicies [][]string 248 248 + var removePolicies [][]string 249 249 + for _, p := range policies { 250 250 + removePolicies = append(removePolicies, p) 251 251 + newPolicy := make([]string, len(p)) 252 252 + copy(newPolicy, p) 253 253 + newPolicy[2] = repoDid 254 254 + addPolicies = append(addPolicies, newPolicy) 255 255 + } 256 256 + 257 257 + if len(addPolicies) > 0 { 258 258 + if _, addErr := e.E.AddPolicies(addPolicies); addErr != nil { 259 259 + return fmt.Errorf("adding new policies: %w", addErr) 260 260 + } 261 261 + } 262 262 + 263 263 + if len(removePolicies) > 0 { 264 264 + if _, rmErr := e.E.RemovePolicies(removePolicies); rmErr != nil { 265 265 + return fmt.Errorf("removing old policies: %w", rmErr) 266 266 + } 267 267 + } 268 268 + 269 269 + l.Info("rewrote RBAC policies", "old", oldResource, "new", repoDid, "count", len(policies)) 270 270 + return nil 271 271 + }

+2

knotserver/server.go

··· 88 88 89 89 notifier := notifier.New() 90 90 91 91 + go migrateReposOnStartup(ctx, c, db, e, &notifier, log.SubLogger(logger, "migrate")) 92 92 + 91 93 mux, err := Setup(ctx, c, db, e, jc, &notifier) 92 94 if err != nil { 93 95 return fmt.Errorf("failed to setup server: %w", err)

+1

spindle/db/db.go

··· 18 18 "_journal_mode=WAL", 19 19 "_synchronous=NORMAL", 20 20 "_auto_vacuum=incremental", 21 21 + "_busy_timeout=5000", 21 22 } 22 23 23 24 db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&"))

+11 -10

spindle/stream.go

··· 52 52 } 53 53 }() 54 54 55 55 - defaultCursor := time.Now().UnixNano() 55 55 + var cursor int64 56 56 cursorStr := r.URL.Query().Get("cursor") 57 57 - cursor, err := strconv.ParseInt(cursorStr, 10, 64) 58 58 - if err != nil { 59 59 - l.Error("empty or invalid cursor", "invalidCursor", cursorStr, "default", defaultCursor) 60 60 - } 61 61 - if cursor == 0 { 62 62 - cursor = defaultCursor 57 57 + if cursorStr != "" { 58 58 + cursor, err = strconv.ParseInt(cursorStr, 10, 64) 59 59 + if err != nil { 60 60 + l.Error("invalid cursor, starting from beginning", "invalidCursor", cursorStr) 61 61 + cursor = 0 62 62 + } 63 63 } 64 64 65 65 // complete backfill first before going to live data ··· 239 239 } 240 240 241 241 jsonMsg, err := json.Marshal(map[string]any{ 242 242 - "rkey": event.Rkey, 243 243 - "nsid": event.Nsid, 244 244 - "event": eventJson, 242 242 + "rkey": event.Rkey, 243 243 + "nsid": event.Nsid, 244 244 + "event": eventJson, 245 245 + "created": event.Created, 245 246 }) 246 247 if err != nil { 247 248 s.l.Error("failed to marshal record", "err", err)