appview,knotserver: improve txn handling around repo creation and forking

+8 -2

appview/pages/templates/repo/fork.html

··· 5 5 <p class="text-xl font-bold dark:text-white">Fork {{ .RepoInfo.FullName }}</p> 6 6 </div> 7 7 <div class="p-6 bg-white dark:bg-gray-800 drop-shadow-sm rounded"> 8 - <form hx-post="/{{ .RepoInfo.FullName }}/fork" class="space-y-12" hx-swap="none"> 8 + <form hx-post="/{{ .RepoInfo.FullName }}/fork" class="space-y-12" hx-swap="none" hx-indicator="#spinner"> 9 9 <fieldset class="space-y-3"> 10 10 <legend class="dark:text-white">Select a knot to fork into</legend> 11 11 <div class="space-y-2"> ··· 30 30 </fieldset> 31 31 32 32 <div class="space-y-2"> 33 - <button type="submit" class="btn">fork repo</button> 33 + <button type="submit" class="btn-create flex items-center gap-2"> 34 + {{ i "git-fork" "w-4 h-4" }} 35 + fork repo 36 + <span id="spinner" class="group"> 37 + {{ i "loader-circle" "w-4 h-4 animate-spin hidden group-[.htmx-request]:inline" }} 38 + </span> 39 + </button> 34 40 <div id="repo" class="error"></div> 35 41 </div> 36 42 </form>

+1 -1

appview/pages/templates/repo/new.html

··· 63 63 <button type="submit" class="btn-create flex items-center gap-2"> 64 64 {{ i "book-plus" "w-4 h-4" }} 65 65 create repo 66 - <span id="create-pull-spinner" class="group"> 66 + <span id="spinner" class="group"> 67 67 {{ i "loader-circle" "w-4 h-4 animate-spin hidden group-[.htmx-request]:inline" }} 68 68 </span> 69 69 </button>

+102 -64

appview/repo/repo.go

··· 28 28 "tangled.sh/tangled.sh/core/appview/pages" 29 29 "tangled.sh/tangled.sh/core/appview/pages/markup" 30 30 "tangled.sh/tangled.sh/core/appview/reporesolver" 31 + xrpcclient "tangled.sh/tangled.sh/core/appview/xrpcclient" 31 32 "tangled.sh/tangled.sh/core/eventconsumer" 32 33 "tangled.sh/tangled.sh/core/idresolver" 33 34 "tangled.sh/tangled.sh/core/knotclient" ··· 1453 1454 }) 1454 1455 1455 1456 case http.MethodPost: 1457 + l := rp.logger.With("handler", "ForkRepo") 1456 1458 1457 - knot := r.FormValue("knot") 1458 - if knot == "" { 1459 + targetKnot := r.FormValue("knot") 1460 + if targetKnot == "" { 1459 1461 rp.pages.Notice(w, "repo", "Invalid form submission—missing knot domain.") 1460 1462 return 1461 1463 } 1464 + l = l.With("targetKnot", targetKnot) 1462 1465 1463 - ok, err := rp.enforcer.E.Enforce(user.Did, knot, knot, "repo:create") 1466 + ok, err := rp.enforcer.E.Enforce(user.Did, targetKnot, targetKnot, "repo:create") 1464 1467 if err != nil || !ok { 1465 1468 rp.pages.Notice(w, "repo", "You do not have permission to create a repo in this knot.") 1466 1469 return 1467 1470 } 1468 1471 1469 - forkName := fmt.Sprintf("%s", f.Name) 1470 - 1472 + // choose a name for a fork 1473 + forkName := f.Name 1471 1474 // this check is *only* to see if the forked repo name already exists 1472 1475 // in the user's account. 1473 1476 existingRepo, err := db.GetRepo(rp.db, user.Did, f.Name) ··· 1483 1486 // repo with this name already exists, append random string 1484 1487 forkName = fmt.Sprintf("%s-%s", forkName, randomString(3)) 1485 1488 } 1486 - client, err := rp.oauth.ServiceClient( 1487 - r, 1488 - oauth.WithService(knot), 1489 - oauth.WithLxm(tangled.RepoForkNSID), 1490 - oauth.WithDev(rp.config.Core.Dev), 1491 - ) 1492 - 1493 - if err != nil { 1494 - log.Printf("error creating client for knot server: %v", err) 1495 - rp.pages.Notice(w, "repo", "Failed to connect to knot server.") 1496 - return 1497 - } 1489 + l = l.With("forkName", forkName) 1498 1490 1499 - var uri string 1491 + uri := "https" 1500 1492 if rp.config.Core.Dev { 1501 1493 uri = "http" 1502 - } else { 1503 - uri = "https" 1504 1494 } 1495 + 1505 1496 forkSourceUrl := fmt.Sprintf("%s://%s/%s/%s", uri, f.Knot, f.OwnerDid(), f.Repo.Name) 1497 + l = l.With("cloneUrl", forkSourceUrl) 1498 + 1506 1499 sourceAt := f.RepoAt().String() 1507 1500 1501 + // create an atproto record for this fork 1508 1502 rkey := tid.TID() 1509 1503 repo := &db.Repo{ 1510 1504 Did: user.Did, 1511 1505 Name: forkName, 1512 - Knot: knot, 1506 + Knot: targetKnot, 1513 1507 Rkey: rkey, 1514 1508 Source: sourceAt, 1515 1509 } 1516 1510 1517 - tx, err := rp.db.BeginTx(r.Context(), nil) 1518 - if err != nil { 1519 - log.Println(err) 1520 - rp.pages.Notice(w, "repo", "Failed to save repository information.") 1521 - return 1522 - } 1523 - defer func() { 1524 - tx.Rollback() 1525 - err = rp.enforcer.E.LoadPolicy() 1526 - if err != nil { 1527 - log.Println("failed to rollback policies") 1528 - } 1529 - }() 1530 - 1531 - err = tangled.RepoFork( 1532 - r.Context(), 1533 - client, 1534 - &tangled.RepoFork_Input{ 1535 - Did: user.Did, 1536 - Name: &forkName, 1537 - Source: forkSourceUrl, 1538 - }, 1539 - ) 1540 - 1541 - if err != nil { 1542 - xe, err := xrpcerr.Unmarshal(err.Error()) 1543 - if err != nil { 1544 - log.Println(err) 1545 - rp.pages.Notice(w, "repo", "Failed to create repository on knot server.") 1546 - return 1547 - } 1548 - 1549 - log.Println(xe.Error()) 1550 - rp.pages.Notice(w, "repo", fmt.Sprintf("Failed to create repository on knot server: %s.", xe.Message)) 1551 - return 1552 - } 1553 - 1554 1511 xrpcClient, err := rp.oauth.AuthorizedClient(r) 1555 1512 if err != nil { 1556 - log.Println("failed to get authorized client", err) 1557 - rp.pages.Notice(w, "repo", "Failed to create repository.") 1513 + l.Error("failed to create xrpcclient", "err", err) 1514 + rp.pages.Notice(w, "repo", "Failed to fork repository.") 1558 1515 return 1559 1516 } 1560 1517 ··· 1573 1530 }}, 1574 1531 }) 1575 1532 if err != nil { 1576 - log.Printf("failed to create record: %s", err) 1533 + l.Error("failed to write to PDS", "err", err) 1577 1534 rp.pages.Notice(w, "repo", "Failed to announce repository creation.") 1578 1535 return 1579 1536 } 1580 - log.Println("created repo record: ", atresp.Uri) 1537 + 1538 + aturi := atresp.Uri 1539 + l = l.With("aturi", aturi) 1540 + l.Info("wrote to PDS") 1541 + 1542 + tx, err := rp.db.BeginTx(r.Context(), nil) 1543 + if err != nil { 1544 + l.Info("txn failed", "err", err) 1545 + rp.pages.Notice(w, "repo", "Failed to save repository information.") 1546 + return 1547 + } 1548 + 1549 + // The rollback function reverts a few things on failure: 1550 + // - the pending txn 1551 + // - the ACLs 1552 + // - the atproto record created 1553 + rollback := func() { 1554 + err1 := tx.Rollback() 1555 + err2 := rp.enforcer.E.LoadPolicy() 1556 + err3 := rollbackRecord(context.Background(), aturi, xrpcClient) 1557 + 1558 + // ignore txn complete errors, this is okay 1559 + if errors.Is(err1, sql.ErrTxDone) { 1560 + err1 = nil 1561 + } 1562 + 1563 + if errs := errors.Join(err1, err2, err3); errs != nil { 1564 + l.Error("failed to rollback changes", "errs", errs) 1565 + return 1566 + } 1567 + } 1568 + defer rollback() 1569 + 1570 + client, err := rp.oauth.ServiceClient( 1571 + r, 1572 + oauth.WithService(targetKnot), 1573 + oauth.WithLxm(tangled.RepoCreateNSID), 1574 + oauth.WithDev(rp.config.Core.Dev), 1575 + ) 1576 + if err != nil { 1577 + l.Error("could not create service client", "err", err) 1578 + rp.pages.Notice(w, "repo", "Failed to connect to knot server.") 1579 + return 1580 + } 1581 + 1582 + err = tangled.RepoCreate( 1583 + r.Context(), 1584 + client, 1585 + &tangled.RepoCreate_Input{ 1586 + Rkey: rkey, 1587 + Source: &forkSourceUrl, 1588 + }, 1589 + ) 1590 + if err := xrpcclient.HandleXrpcErr(err); err != nil { 1591 + rp.pages.Notice(w, "repo", err.Error()) 1592 + return 1593 + } 1581 1594 1582 1595 err = db.AddRepo(tx, repo) 1583 1596 if err != nil { ··· 1588 1601 1589 1602 // acls 1590 1603 p, _ := securejoin.SecureJoin(user.Did, forkName) 1591 - err = rp.enforcer.AddRepo(user.Did, knot, p) 1604 + err = rp.enforcer.AddRepo(user.Did, targetKnot, p) 1592 1605 if err != nil { 1593 1606 log.Println(err) 1594 1607 rp.pages.Notice(w, "repo", "Failed to set up repository permissions.") ··· 1609 1622 return 1610 1623 } 1611 1624 1625 + // reset the ATURI because the transaction completed successfully 1626 + aturi = "" 1627 + 1628 + rp.notifier.NewRepo(r.Context(), repo) 1612 1629 rp.pages.HxLocation(w, fmt.Sprintf("/@%s/%s", user.Handle, forkName)) 1613 - return 1630 + } 1631 + } 1632 + 1633 + // this is used to rollback changes made to the PDS 1634 + // 1635 + // it is a no-op if the provided ATURI is empty 1636 + func rollbackRecord(ctx context.Context, aturi string, xrpcc *xrpcclient.Client) error { 1637 + if aturi == "" { 1638 + return nil 1614 1639 } 1640 + 1641 + parsed := syntax.ATURI(aturi) 1642 + 1643 + collection := parsed.Collection().String() 1644 + repo := parsed.Authority().String() 1645 + rkey := parsed.RecordKey().String() 1646 + 1647 + _, err := xrpcc.RepoDeleteRecord(ctx, &comatproto.RepoDeleteRecord_Input{ 1648 + Collection: collection, 1649 + Repo: repo, 1650 + Rkey: rkey, 1651 + }) 1652 + return err 1615 1653 } 1616 1654 1617 1655 func (rp *Repo) RepoCompareNew(w http.ResponseWriter, r *http.Request) {

+88 -37

appview/state/state.go

··· 2 2 3 3 import ( 4 4 "context" 5 + "database/sql" 6 + "errors" 5 7 "fmt" 6 8 "log" 7 9 "log/slog" ··· 10 12 "time" 11 13 12 14 comatproto "github.com/bluesky-social/indigo/api/atproto" 15 + "github.com/bluesky-social/indigo/atproto/syntax" 13 16 lexutil "github.com/bluesky-social/indigo/lex/util" 14 17 securejoin "github.com/cyphar/filepath-securejoin" 15 18 "github.com/go-chi/chi/v5" ··· 25 28 "tangled.sh/tangled.sh/core/appview/pages" 26 29 posthogService "tangled.sh/tangled.sh/core/appview/posthog" 27 30 "tangled.sh/tangled.sh/core/appview/reporesolver" 31 + xrpcclient "tangled.sh/tangled.sh/core/appview/xrpcclient" 28 32 "tangled.sh/tangled.sh/core/eventconsumer" 29 33 "tangled.sh/tangled.sh/core/idresolver" 30 34 "tangled.sh/tangled.sh/core/jetstream" ··· 48 52 repoResolver *reporesolver.RepoResolver 49 53 knotstream *eventconsumer.Consumer 50 54 spindlestream *eventconsumer.Consumer 55 + logger *slog.Logger 51 56 } 52 57 53 58 func Make(ctx context.Context, config *config.Config) (*State, error) { ··· 152 157 repoResolver, 153 158 knotstream, 154 159 spindlestream, 160 + slog.Default(), 155 161 } 156 162 157 163 return state, nil ··· 291 297 }) 292 298 293 299 case http.MethodPost: 300 + l := s.logger.With("handler", "NewRepo") 301 + 294 302 user := s.oauth.GetUser(r) 303 + l = l.With("did", user.Did) 304 + l = l.With("handle", user.Handle) 295 305 306 + // form validation 296 307 domain := r.FormValue("domain") 297 308 if domain == "" { 298 309 s.pages.Notice(w, "repo", "Invalid form submission—missing knot domain.") 299 310 return 300 311 } 312 + l = l.With("knot", domain) 301 313 302 314 repoName := r.FormValue("name") 303 315 if repoName == "" { ··· 309 321 s.pages.Notice(w, "repo", err.Error()) 310 322 return 311 323 } 312 - 313 324 repoName = stripGitExt(repoName) 325 + l = l.With("repoName", repoName) 314 326 315 327 defaultBranch := r.FormValue("branch") 316 328 if defaultBranch == "" { 317 329 defaultBranch = "main" 318 330 } 331 + l = l.With("defaultBranch", defaultBranch) 319 332 320 333 description := r.FormValue("description") 321 334 335 + // ACL validation 322 336 ok, err := s.enforcer.E.Enforce(user.Did, domain, domain, "repo:create") 323 337 if err != nil || !ok { 338 + l.Info("unauthorized") 324 339 s.pages.Notice(w, "repo", "You do not have permission to create a repo in this knot.") 325 340 return 326 341 } 327 342 343 + // Check for existing repos 328 344 existingRepo, err := db.GetRepo(s.db, user.Did, repoName) 329 345 if err == nil && existingRepo != nil { 330 346 l.Info("repo exists") ··· 332 348 return 333 349 } 334 350 335 - client, err := s.oauth.ServiceClient( 336 - r, 337 - oauth.WithService(domain), 338 - oauth.WithLxm(tangled.RepoCreateNSID), 339 - oauth.WithDev(s.config.Core.Dev), 340 - ) 341 - 342 - if err != nil { 343 - s.pages.Notice(w, "repo", "Failed to connect to knot server.") 344 - return 345 - } 346 - 351 + // create atproto record for this repo 347 352 rkey := tid.TID() 348 353 repo := &db.Repo{ 349 354 Did: user.Did, ··· 355 360 356 361 xrpcClient, err := s.oauth.AuthorizedClient(r) 357 362 if err != nil { 363 + l.Info("PDS write failed", "err", err) 358 364 s.pages.Notice(w, "repo", "Failed to write record to PDS.") 359 365 return 360 366 } ··· 373 379 }}, 374 380 }) 375 381 if err != nil { 376 - log.Printf("failed to create record: %s", err) 382 + l.Info("PDS write failed", "err", err) 377 383 s.pages.Notice(w, "repo", "Failed to announce repository creation.") 378 384 return 379 385 } 380 - log.Println("created repo record: ", atresp.Uri) 386 + 387 + aturi := atresp.Uri 388 + l = l.With("aturi", aturi) 389 + l.Info("wrote to PDS") 381 390 382 391 tx, err := s.db.BeginTx(r.Context(), nil) 383 392 if err != nil { 384 - log.Println(err) 393 + l.Info("txn failed", "err", err) 385 394 s.pages.Notice(w, "repo", "Failed to save repository information.") 386 395 return 387 396 } 388 - defer func() { 389 - tx.Rollback() 390 - err = s.enforcer.E.LoadPolicy() 391 - if err != nil { 392 - log.Println("failed to rollback policies") 397 + 398 + // The rollback function reverts a few things on failure: 399 + // - the pending txn 400 + // - the ACLs 401 + // - the atproto record created 402 + rollback := func() { 403 + err1 := tx.Rollback() 404 + err2 := s.enforcer.E.LoadPolicy() 405 + err3 := rollbackRecord(context.Background(), aturi, xrpcClient) 406 + 407 + // ignore txn complete errors, this is okay 408 + if errors.Is(err1, sql.ErrTxDone) { 409 + err1 = nil 410 + } 411 + 412 + if errs := errors.Join(err1, err2, err3); errs != nil { 413 + l.Error("failed to rollback changes", "errs", errs) 414 + return 393 415 } 394 - }() 416 + } 417 + defer rollback() 418 + 419 + client, err := s.oauth.ServiceClient( 420 + r, 421 + oauth.WithService(domain), 422 + oauth.WithLxm(tangled.RepoCreateNSID), 423 + oauth.WithDev(s.config.Core.Dev), 424 + ) 425 + if err != nil { 426 + l.Error("service auth failed", "err", err) 427 + s.pages.Notice(w, "repo", "Failed to reach PDS.") 428 + return 429 + } 395 430 396 431 xe := tangled.RepoCreate( 397 432 r.Context(), ··· 401 436 }, 402 437 ) 403 438 if err != nil { 404 - xe, err := xrpcerr.Unmarshal(err.Error()) 405 - if err != nil { 406 - log.Println(err) 407 - s.pages.Notice(w, "repo", "Failed to create repository on knot server.") 408 - return 409 - } 410 - 411 - log.Println(xe.Error()) 412 - s.pages.Notice(w, "repo", fmt.Sprintf("Failed to create repository on knot server: %s.", xe.Message)) 439 + l.Error("xrpc request failed", "err", err) 440 + s.pages.Notice(w, "repo", fmt.Sprintf("Failed to create repository on knot server: %s.", err.Error())) 413 441 return 414 442 } 415 443 416 444 err = db.AddRepo(tx, repo) 417 445 if err != nil { 418 - log.Println(err) 446 + l.Error("db write failed", "err", err) 419 447 s.pages.Notice(w, "repo", "Failed to save repository information.") 420 448 return 421 449 } ··· 424 452 p, _ := securejoin.SecureJoin(user.Did, repoName) 425 453 err = s.enforcer.AddRepo(user.Did, domain, p) 426 454 if err != nil { 427 - log.Println(err) 455 + l.Error("acl setup failed", "err", err) 428 456 s.pages.Notice(w, "repo", "Failed to set up repository permissions.") 429 457 return 430 458 } 431 459 432 460 err = tx.Commit() 433 461 if err != nil { 434 - log.Println("failed to commit changes", err) 462 + l.Error("txn commit failed", "err", err) 435 463 http.Error(w, err.Error(), http.StatusInternalServerError) 436 464 return 437 465 } 438 466 439 467 err = s.enforcer.E.SavePolicy() 440 468 if err != nil { 441 - log.Println("failed to update ACLs", err) 469 + l.Error("acl save failed", "err", err) 442 470 http.Error(w, err.Error(), http.StatusInternalServerError) 443 471 return 444 472 } 445 473 474 + // reset the ATURI because the transaction completed successfully 475 + aturi = "" 476 + 446 477 s.notifier.NewRepo(r.Context(), repo) 478 + s.pages.HxLocation(w, fmt.Sprintf("/@%s/%s", user.Handle, repoName)) 479 + } 480 + } 447 481 448 - s.pages.HxLocation(w, fmt.Sprintf("/@%s/%s", user.Handle, repoName)) 449 - return 482 + // this is used to rollback changes made to the PDS 483 + // 484 + // it is a no-op if the provided ATURI is empty 485 + func rollbackRecord(ctx context.Context, aturi string, xrpcc *xrpcclient.Client) error { 486 + if aturi == "" { 487 + return nil 450 488 } 489 + 490 + parsed := syntax.ATURI(aturi) 491 + 492 + collection := parsed.Collection().String() 493 + repo := parsed.Authority().String() 494 + rkey := parsed.RecordKey().String() 495 + 496 + _, err := xrpcc.RepoDeleteRecord(ctx, &comatproto.RepoDeleteRecord_Input{ 497 + Collection: collection, 498 + Repo: repo, 499 + Rkey: rkey, 500 + }) 501 + return err 451 502 }

+8 -5

knotserver/xrpc/router.go knotserver/xrpc/xrpc.go

··· 31 31 32 32 func (x *Xrpc) Router() http.Handler { 33 33 r := chi.NewRouter() 34 + 34 35 r.Group(func(r chi.Router) { 35 36 r.Use(x.ServiceAuth.VerifyServiceAuth) 36 37 37 38 r.Post("/"+tangled.RepoSetDefaultBranchNSID, x.SetDefaultBranch) 38 39 r.Post("/"+tangled.RepoCreateNSID, x.CreateRepo) 39 - r.Post("/"+tangled.RepoDeleteNSID, x.DeleteRepo) 40 - r.Post("/"+tangled.RepoForkNSID, x.ForkRepo) 41 40 r.Post("/"+tangled.RepoForkStatusNSID, x.ForkStatus) 42 41 r.Post("/"+tangled.RepoForkSyncNSID, x.ForkSync) 43 - 44 42 r.Post("/"+tangled.RepoHiddenRefNSID, x.HiddenRef) 45 - 46 43 r.Post("/"+tangled.RepoMergeNSID, x.Merge) 47 - r.Post("/"+tangled.RepoMergeCheckNSID, x.MergeCheck) 48 44 }) 45 + 46 + // merge check is an open endpoint 47 + // 48 + // TODO: should we constrain this more? 49 + // - we can calculate on PR submit/resubmit/gitRefUpdate etc. 50 + // - use ETags on clients to keep requests to a minimum 51 + r.Post("/"+tangled.RepoMergeCheckNSID, x.MergeCheck) 49 52 return r 50 53 } 51 54

Configure Feed

Configure Feed