Compare commits

..

7 Commits

7 changed files with 479 additions and 115 deletions
+4 -5
View File
@@ -533,8 +533,8 @@ func (s *Store) DeleteBlockAttachment(ctx context.Context, boardID, blockID, att
} }
err = scanBlockAttachment( err = scanBlockAttachment(
tx.QueryRow(`SELECT`+blockAttachmentSelectCols+blockAttachmentSelectFrom+` tx.QueryRow(`SELECT`+blockAttachmentSelectCols+blockAttachmentSelectFrom+`
JOIN boards brd ON brd.id = bl.board_id JOIN boards brd ON brd.id = bl.board_id
WHERE brd.public_id = ? AND bl.public_id = ? AND bl.delete_at = 0 AND ba.public_id = ? AND ba.delete_at = 0`, WHERE brd.public_id = ? AND bl.public_id = ? AND bl.delete_at = 0 AND ba.public_id = ? AND ba.delete_at = 0`,
boardID, blockID, attachmentID), boardID, blockID, attachmentID),
&item) &item)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
@@ -570,10 +570,9 @@ func (s *Store) DeleteBlockAttachment(ctx context.Context, boardID, blockID, att
return item, ErrBlockAttachmentNotFound return item, ErrBlockAttachmentNotFound
} }
item.DeleteAt = now item.DeleteAt = now
err = commitIfOwned(tx, owned) err = commitIfOwned(tx, owned)
if err != nil { if err != nil { return item, err }
return item, err
}
return item, nil return item, nil
} }
+3
View File
@@ -8,6 +8,9 @@ import "os"
import "path/filepath" import "path/filepath"
import "sort" import "sort"
import "strings" import "strings"
import "time"
const IndependentDBOperationTimeout = 10 * time.Second
type sqlExecutor interface { type sqlExecutor interface {
Exec(query string, args ...any) (sql.Result, error) Exec(query string, args ...any) (sql.Result, error)
+28 -1
View File
@@ -16,6 +16,7 @@ import "codit/config"
import "codit/internal/db" import "codit/internal/db"
import "codit/internal/middleware" import "codit/internal/middleware"
import "codit/internal/models" import "codit/internal/models"
import "codit/internal/repolock"
import "codit/internal/util" import "codit/internal/util"
import codit_logger "codit/logger" import codit_logger "codit/logger"
@@ -27,17 +28,19 @@ type HTTPServer struct {
auth AuthFunc auth AuthFunc
logger codit_logger.Logger logger codit_logger.Logger
prefix string prefix string
locks *repolock.Manager
} }
const logIDDocker string = "docker" const logIDDocker string = "docker"
func NewHTTPServer(store *db.Store, baseDir string, auth AuthFunc, logger codit_logger.Logger, prefix string) *HTTPServer { func NewHTTPServer(store *db.Store, baseDir string, auth AuthFunc, logger codit_logger.Logger, prefix string, locks *repolock.Manager) *HTTPServer {
return &HTTPServer{ return &HTTPServer{
store: store, store: store,
baseDir: baseDir, baseDir: baseDir,
auth: auth, auth: auth,
logger: logger, logger: logger,
prefix: config.NormalizeHTTPPrefix(prefix, "/v2"), prefix: config.NormalizeHTTPPrefix(prefix, "/v2"),
locks: locks,
} }
} }
@@ -107,6 +110,8 @@ func (s *HTTPServer) handle(w http.ResponseWriter, r *http.Request) *http.Reques
var project models.Project var project models.Project
var imageName string var imageName string
var imagePath string var imagePath string
var unlock func()
var locked bool
var err error var err error
path = r.URL.Path path = r.URL.Path
store = s.requestStore(r) store = s.requestStore(r)
@@ -128,6 +133,14 @@ func (s *HTTPServer) handle(w http.ResponseWriter, r *http.Request) *http.Reques
w.WriteHeader(http.StatusNotFound) w.WriteHeader(http.StatusNotFound)
return r return r
} }
if s.isWriteRequest(action, r.Method) {
unlock, locked = s.tryLockRepo(repo.ID)
if !locked {
w.WriteHeader(http.StatusConflict)
return r
}
defer unlock()
}
r = WithRequestLogInfo(r, RequestLogInfo{ r = WithRequestLogInfo(r, RequestLogInfo{
OrgPath: path, OrgPath: path,
ProjectID: project.ID, ProjectID: project.ID,
@@ -153,6 +166,20 @@ func (s *HTTPServer) handle(w http.ResponseWriter, r *http.Request) *http.Reques
return r return r
} }
func (s *HTTPServer) tryLockRepo(repoID string) (func(), bool) {
if s.locks == nil { return func() {}, true }
return s.locks.TryLock(repoID)
}
func (s *HTTPServer) isWriteRequest(action string, method string) bool {
if action == "manifest" && method == http.MethodPut { return true }
if action == "upload_start" && method == http.MethodPost { return true }
if action == "upload" {
if method == http.MethodPatch || method == http.MethodPut || method == http.MethodDelete { return true }
}
return false
}
func (s *HTTPServer) requestStore(r *http.Request) *db.Store { func (s *HTTPServer) requestStore(r *http.Request) *db.Store {
var store *db.Store var store *db.Store
var ok bool var ok bool
+275 -62
View File
@@ -28,6 +28,7 @@ import "codit/internal/git"
import "codit/internal/middleware" import "codit/internal/middleware"
import "codit/internal/models" import "codit/internal/models"
import "codit/internal/rpm" import "codit/internal/rpm"
import "codit/internal/repolock"
import "codit/internal/storage" import "codit/internal/storage"
import "codit/internal/util" import "codit/internal/util"
import codit_logger "codit/logger" import codit_logger "codit/logger"
@@ -46,6 +47,7 @@ type APIOptions struct {
Logger codit_logger.Logger Logger codit_logger.Logger
OnTLSListenersChanged func() OnTLSListenersChanged func()
OnTLSListenerRuntimeStatus func() map[string]int OnTLSListenerRuntimeStatus func() map[string]int
RepoLocks *repolock.Manager
} }
type API struct { type API struct {
@@ -68,6 +70,7 @@ type API struct {
SSHPreparedSessionStore *SSHPreparedSessionStore SSHPreparedSessionStore *SSHPreparedSessionStore
OnTLSListenersChanged func() OnTLSListenersChanged func()
OnTLSListenerRuntimeStatus func() map[string]int OnTLSListenerRuntimeStatus func() map[string]int
RepoLocks *repolock.Manager
} }
type appInfoResponse struct { type appInfoResponse struct {
@@ -104,10 +107,28 @@ func NewAPI(options APIOptions) *API {
SSHPreparedSessionStore: NewSSHPreparedSessionStore(), SSHPreparedSessionStore: NewSSHPreparedSessionStore(),
OnTLSListenersChanged: options.OnTLSListenersChanged, OnTLSListenersChanged: options.OnTLSListenersChanged,
OnTLSListenerRuntimeStatus: options.OnTLSListenerRuntimeStatus, OnTLSListenerRuntimeStatus: options.OnTLSListenerRuntimeStatus,
RepoLocks: options.RepoLocks,
} }
if api.RepoLocks == nil { api.RepoLocks = repolock.NewManager() }
return api return api
} }
func (api *API) lockRepo(repoID string) func() {
return api.RepoLocks.Lock(repoID)
}
func (api *API) lockRepos(repoIDs []string) func() {
return api.RepoLocks.LockMany(repoIDs)
}
func (api *API) tryLockRepo(repoID string) (func(), bool) {
return api.RepoLocks.TryLock(repoID)
}
func (api *API) tryLockRepos(repoIDs []string) (func(), bool) {
return api.RepoLocks.TryLockMany(repoIDs)
}
func (api *API) serverId() string { func (api *API) serverId() string {
return config.NormalizeServerId(api.ServerId) return config.NormalizeServerId(api.ServerId)
} }
@@ -890,6 +911,7 @@ func (api *API) UploadMyAvatar(w http.ResponseWriter, r *http.Request, _ map[str
var copied int64 var copied int64
var ts string var ts string
var closeErr error var closeErr error
var txStore *db.Store
ctxUser, ok = middleware.UserFromContext(r.Context()) ctxUser, ok = middleware.UserFromContext(r.Context())
if !ok { if !ok {
@@ -930,9 +952,7 @@ func (api *API) UploadMyAvatar(w http.ResponseWriter, r *http.Request, _ map[str
} }
copied, err = io.Copy(out, file) copied, err = io.Copy(out, file)
closeErr = out.Close() closeErr = out.Close()
if err == nil { if err == nil { err = closeErr }
err = closeErr
}
if err != nil { if err != nil {
_ = os.Remove(tempPath) _ = os.Remove(tempPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
@@ -949,21 +969,44 @@ func (api *API) UploadMyAvatar(w http.ResponseWriter, r *http.Request, _ map[str
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
oldAvatar, oldErr = api.store(r).GetUserAvatar(ctxUser.ID)
err = api.store(r).UpdateUserAvatar(ctxUser.ID, finalPath, contentType) txStore, err = api.Store.BeginImmediateStore(r.Context())
if err != nil { if err != nil {
_ = os.Remove(finalPath) _ = os.Remove(finalPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
oldAvatar, oldErr = txStore.GetUserAvatar(ctxUser.ID)
if oldErr != nil && oldErr != sql.ErrNoRows {
_ = txStore.Rollback()
_ = os.Remove(finalPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, oldErr.Error())
return
}
err = txStore.UpdateUserAvatar(ctxUser.ID, finalPath, contentType)
if err != nil {
_ = txStore.Rollback()
_ = os.Remove(finalPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
user, err = txStore.GetUserByID(ctxUser.ID)
if err != nil {
_ = txStore.Rollback()
_ = os.Remove(finalPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
err = txStore.Commit()
if err != nil {
_ = txStore.Rollback()
_ = os.Remove(finalPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
if oldErr == nil && oldAvatar.StoragePath != "" && oldAvatar.StoragePath != finalPath { if oldErr == nil && oldAvatar.StoragePath != "" && oldAvatar.StoragePath != finalPath {
_ = os.Remove(oldAvatar.StoragePath) _ = os.Remove(oldAvatar.StoragePath)
} }
user, err = api.store(r).GetUserByID(ctxUser.ID)
if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
WriteJSON(w, http.StatusOK, user) WriteJSON(w, http.StatusOK, user)
} }
@@ -973,30 +1016,45 @@ func (api *API) DeleteMyAvatar(w http.ResponseWriter, r *http.Request, _ map[str
var avatar models.UserAvatar var avatar models.UserAvatar
var ok bool var ok bool
var err error var err error
var txStore *db.Store
ctxUser, ok = middleware.UserFromContext(r.Context()) ctxUser, ok = middleware.UserFromContext(r.Context())
if !ok { if !ok {
w.WriteHeader(http.StatusUnauthorized) w.WriteHeader(http.StatusUnauthorized)
return return
} }
avatar, err = api.store(r).GetUserAvatar(ctxUser.ID) txStore, err = api.Store.BeginImmediateStore(r.Context())
if err != nil && err != sql.ErrNoRows { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
err = api.store(r).ClearUserAvatar(ctxUser.ID) avatar, err = txStore.GetUserAvatar(ctxUser.ID)
if err != nil && err != sql.ErrNoRows {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
err = txStore.ClearUserAvatar(ctxUser.ID)
if err != nil { if err != nil {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
user, err = txStore.GetUserByID(ctxUser.ID)
if err != nil {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
err = txStore.Commit()
if err != nil {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
if avatar.StoragePath != "" { if avatar.StoragePath != "" {
_ = os.Remove(avatar.StoragePath) _ = os.Remove(avatar.StoragePath)
} }
user, err = api.store(r).GetUserByID(ctxUser.ID)
if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
WriteJSON(w, http.StatusOK, user) WriteJSON(w, http.StatusOK, user)
} }
@@ -2212,6 +2270,9 @@ func (api *API) DeleteProject(w http.ResponseWriter, r *http.Request, params map
var sourcePaths []string var sourcePaths []string
var temp string var temp string
var txStore *db.Store var txStore *db.Store
var repoIDs []string
var unlock func()
var locked bool
if !api.requireProjectRole(w, r, params["id"], models.RoleAdmin) { return } if !api.requireProjectRole(w, r, params["id"], models.RoleAdmin) { return }
@@ -2220,15 +2281,28 @@ func (api *API) DeleteProject(w http.ResponseWriter, r *http.Request, params map
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
for i = 0; i < len(repos); i++ {
repoIDs = append(repoIDs, repos[i].ID)
}
unlock, locked = api.tryLockRepos(repoIDs)
if !locked {
WriteJSONWithErrorReason(w, r, http.StatusConflict, "repository is busy")
return
}
defer unlock()
currentRepos, err = api.store(r).ListReposOwned(params["id"])
if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
if !sameRepoIDSet(repos, currentRepos) {
WriteJSONWithErrorReason(w, r, http.StatusConflict, "project repositories changed while deletion was starting")
return
}
repos = currentRepos
// Prevent deletion if a repo row is still in the provisioning window. // Keep the explicit path check as a guard against incomplete or corrupted
// This can happen due to race condition with CreateRepo. // repo rows. Normal in-process repo creation is serialized by repo locks.
// CreateRepo has two transactions:
// 1. Short DB transaction creates repo row with path="".
// 2. Transaction commits.
// 3. Handler initializes filesystem/git storage.
// 4. Second short DB transaction updates the repo row with the final path.
// If DeleteProject comes while 3 is ongoing, there is a risk.
for i = 0; i < len(repos); i++ { for i = 0; i < len(repos); i++ {
if repos[i].Path == "" { if repos[i].Path == "" {
WriteJSON(w, http.StatusConflict, map[string]string{"error": "repository storage is not ready", "repo_id": repos[i].ID, "repo_name": repos[i].Name}) WriteJSON(w, http.StatusConflict, map[string]string{"error": "repository storage is not ready", "repo_id": repos[i].ID, "repo_name": repos[i].Name})
@@ -2710,6 +2784,7 @@ func (api *API) CreateRepo(w http.ResponseWriter, r *http.Request, params map[st
var dockerURL string var dockerURL string
var ok bool var ok bool
var err error var err error
var unlock func()
if !api.requireProjectRole(w, r, params["projectId"], models.RoleWriter) { return } if !api.requireProjectRole(w, r, params["projectId"], models.RoleWriter) { return }
@@ -2738,6 +2813,8 @@ func (api *API) CreateRepo(w http.ResponseWriter, r *http.Request, params map[st
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, "failed to create repo") WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, "failed to create repo")
return return
} }
unlock = api.lockRepo(repoID)
defer unlock()
repo = models.Repo{ repo = models.Repo{
ID: repoID, ID: repoID,
@@ -3066,6 +3143,8 @@ func (api *API) DeleteRepo(w http.ResponseWriter, r *http.Request, params map[st
var rpmURL string var rpmURL string
var dockerURL string var dockerURL string
var err error var err error
var unlock func()
var locked bool
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
@@ -3074,6 +3153,12 @@ func (api *API) DeleteRepo(w http.ResponseWriter, r *http.Request, params map[st
} }
if !api.requireProjectRole(w, r, repo.ProjectID, models.RoleWriter) { return } if !api.requireProjectRole(w, r, repo.ProjectID, models.RoleWriter) { return }
unlock, locked = api.tryLockRepo(repo.ID)
if !locked {
WriteJSONWithErrorReason(w, r, http.StatusConflict, "repository is busy")
return
}
defer unlock()
txStore, err = api.Store.BeginImmediateStore(r.Context()) txStore, err = api.Store.BeginImmediateStore(r.Context())
if err != nil { if err != nil {
@@ -3081,7 +3166,8 @@ func (api *API) DeleteRepo(w http.ResponseWriter, r *http.Request, params map[st
return return
} }
// re-read to avoid race condition with CreateRepo which has uses two transactions // Re-read under the delete transaction to validate current DB state before
// moving repository storage.
repo, err = txStore.GetRepo(params["id"]) repo, err = txStore.GetRepo(params["id"])
if err != nil { if err != nil {
_ = txStore.Rollback() _ = txStore.Rollback()
@@ -3230,6 +3316,25 @@ func sameRepoStorageSet(a []models.Repo, b []models.Repo) bool {
return true return true
} }
func sameRepoIDSet(a []models.Repo, b []models.Repo) bool {
var seen map[string]bool
var i int
if len(a) != len(b) {
return false
}
seen = make(map[string]bool, len(a))
for i = 0; i < len(a); i++ {
seen[a[i].ID] = true
}
for i = 0; i < len(b); i++ {
if !seen[b[i].ID] {
return false
}
}
return true
}
func moveToTrash(path string) (string, error) { func moveToTrash(path string) (string, error) {
var dir string var dir string
var base string var base string
@@ -3372,6 +3477,8 @@ func (api *API) RepoSetDefaultBranch(w http.ResponseWriter, r *http.Request, par
var repo models.Repo var repo models.Repo
var err error var err error
var req repoBranchRequest var req repoBranchRequest
var unlock func()
if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) { if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) {
return return
} }
@@ -3380,6 +3487,8 @@ func (api *API) RepoSetDefaultBranch(w http.ResponseWriter, r *http.Request, par
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil || strings.TrimSpace(req.Name) == "" { if err != nil || strings.TrimSpace(req.Name) == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "branch name required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "branch name required")
@@ -3398,6 +3507,8 @@ func (api *API) RepoDeleteBranch(w http.ResponseWriter, r *http.Request, params
var err error var err error
var req repoBranchRequest var req repoBranchRequest
var name string var name string
var unlock func()
if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) { if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) {
return return
} }
@@ -3406,6 +3517,8 @@ func (api *API) RepoDeleteBranch(w http.ResponseWriter, r *http.Request, params
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -3430,6 +3543,8 @@ func (api *API) RepoRenameBranch(w http.ResponseWriter, r *http.Request, params
var req repoBranchRenameRequest var req repoBranchRenameRequest
var from string var from string
var to string var to string
var unlock func()
if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) { if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) {
return return
} }
@@ -3438,6 +3553,8 @@ func (api *API) RepoRenameBranch(w http.ResponseWriter, r *http.Request, params
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -3465,6 +3582,8 @@ func (api *API) RepoCreateBranch(w http.ResponseWriter, r *http.Request, params
var repo models.Repo var repo models.Repo
var err error var err error
var req repoBranchCreateRequest var req repoBranchCreateRequest
var unlock func()
if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) { if !api.requireRepoRole(w, r, params["id"], models.RoleAdmin) {
return return
} }
@@ -3473,6 +3592,8 @@ func (api *API) RepoCreateBranch(w http.ResponseWriter, r *http.Request, params
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -3905,6 +4026,8 @@ func (api *API) RepoRPMCreateSubdir(w http.ResponseWriter, r *http.Request, para
var allowDelete bool var allowDelete bool
var tlsInsecureSkipVerify bool var tlsInsecureSkipVerify bool
var syncIntervalSec int64 var syncIntervalSec int64
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -3917,6 +4040,8 @@ func (api *API) RepoRPMCreateSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -3996,6 +4121,14 @@ func (api *API) RepoRPMCreateSubdir(w http.ResponseWriter, r *http.Request, para
} }
} }
fullPath = filepath.Join(repo.Path, parentPath, name) fullPath = filepath.Join(repo.Path, parentPath, name)
_, err = os.Stat(fullPath)
if err == nil {
WriteJSONWithErrorReason(w, r, http.StatusConflict, "path already exists")
return
} else if err != nil && !os.IsNotExist(err) {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
err = os.MkdirAll(fullPath, 0o755) err = os.MkdirAll(fullPath, 0o755)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
@@ -4005,6 +4138,7 @@ func (api *API) RepoRPMCreateSubdir(w http.ResponseWriter, r *http.Request, para
repodataPath = filepath.Join(fullPath, "repodata") repodataPath = filepath.Join(fullPath, "repodata")
err = os.MkdirAll(repodataPath, 0o755) err = os.MkdirAll(repodataPath, 0o755)
if err != nil { if err != nil {
_ = os.RemoveAll(fullPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
@@ -4023,6 +4157,7 @@ func (api *API) RepoRPMCreateSubdir(w http.ResponseWriter, r *http.Request, para
} }
err = api.store(r).UpsertRPMRepoDir(dirConfig) err = api.store(r).UpsertRPMRepoDir(dirConfig)
if err != nil { if err != nil {
_ = os.RemoveAll(fullPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
@@ -4036,6 +4171,7 @@ func (api *API) RepoRPMGetSubdir(w http.ResponseWriter, r *http.Request, params
var normalizedPath string var normalizedPath string
var config models.RPMRepoDir var config models.RPMRepoDir
var err error var err error
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -4076,18 +4212,24 @@ func (api *API) RepoRPMSyncSubdir(w http.ResponseWriter, r *http.Request, params
var normalizedPath string var normalizedPath string
var config models.RPMRepoDir var config models.RPMRepoDir
var err error var err error
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) {
return if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) { return }
}
if repo.Type != models.RepoTypeRPM { if repo.Type != models.RepoTypeRPM {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -4115,6 +4257,7 @@ func (api *API) RepoRPMSyncSubdir(w http.ResponseWriter, r *http.Request, params
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "mirror sync is suspended") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "mirror sync is suspended")
return return
} }
err = api.store(r).MarkRPMMirrorTaskDirty(repo.ID, normalizedPath) err = api.store(r).MarkRPMMirrorTaskDirty(repo.ID, normalizedPath)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
@@ -4138,18 +4281,24 @@ func (api *API) RepoRPMRebuildSubdirMetadata(w http.ResponseWriter, r *http.Requ
var fullPath string var fullPath string
var repodataPath string var repodataPath string
var err error var err error
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) {
return if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) { return }
}
if repo.Type != models.RepoTypeRPM { if repo.Type != models.RepoTypeRPM {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -4186,6 +4335,8 @@ func (api *API) RepoRPMCancelSubdirSync(w http.ResponseWriter, r *http.Request,
var config models.RPMRepoDir var config models.RPMRepoDir
var canceled bool var canceled bool
var err error var err error
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -4198,6 +4349,8 @@ func (api *API) RepoRPMCancelSubdirSync(w http.ResponseWriter, r *http.Request,
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -4244,6 +4397,8 @@ func (api *API) repoRPMSetSyncEnabled(w http.ResponseWriter, r *http.Request, pa
var config models.RPMRepoDir var config models.RPMRepoDir
var cancelRequested bool var cancelRequested bool
var err error var err error
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -4256,6 +4411,8 @@ func (api *API) repoRPMSetSyncEnabled(w http.ResponseWriter, r *http.Request, pa
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -4390,6 +4547,11 @@ func (api *API) RepoRPMDeleteSubdir(w http.ResponseWriter, r *http.Request, para
var relPathClean string var relPathClean string
var parentPath string var parentPath string
var repodataPath string var repodataPath string
var ctx context.Context
var cancel context.CancelFunc
var txStore *db.Store
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -4402,6 +4564,8 @@ func (api *API) RepoRPMDeleteSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -4495,17 +4659,32 @@ func (api *API) RepoRPMDeleteSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
err = api.store(r).DeleteRPMRepoDirSubtree(repo.ID, relPathClean)
ctx, cancel = context.WithTimeout(context.Background(), db.IndependentDBOperationTimeout)
defer cancel()
txStore, err = api.Store.BeginImmediateStore(ctx)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
err = txStore.DeleteRPMRepoDirSubtree(repo.ID, relPathClean)
if err != nil {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
err = txStore.Commit()
if err != nil {
_ = txStore.Rollback()
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
parentPath = filepath.Dir(fullPath) parentPath = filepath.Dir(fullPath)
repodataPath = filepath.Join(parentPath, "repodata") repodataPath = filepath.Join(parentPath, "repodata")
_, err = os.Stat(repodataPath) _, err = os.Stat(repodataPath)
if err == nil && api.RpmMeta != nil { if err == nil && api.RpmMeta != nil { api.RpmMeta.Schedule(parentPath) }
api.RpmMeta.Schedule(parentPath)
}
WriteJSON(w, http.StatusOK, map[string]string{"status": "ok"}) WriteJSON(w, http.StatusOK, map[string]string{"status": "ok"})
} }
@@ -4545,18 +4724,25 @@ func (api *API) RepoRPMRenameSubdir(w http.ResponseWriter, r *http.Request, para
var busy bool var busy bool
var busyPath string var busyPath string
var busyReason string var busyReason string
var ctx context.Context
var cancel context.CancelFunc
var txStore *db.Store
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
return return
} }
if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) {
return if !api.requireRepoRole(w, r, repo.ID, models.RoleWriter) { return }
}
if repo.Type != models.RepoTypeRPM { if repo.Type != models.RepoTypeRPM {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -4628,13 +4814,13 @@ func (api *API) RepoRPMRenameSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path is not a directory") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path is not a directory")
return return
} }
parentRel = filepath.Dir(filepath.FromSlash(relPathClean)) parentRel = filepath.Dir(filepath.FromSlash(relPathClean))
if parentRel == "." { if parentRel == "." { parentRel = "" }
parentRel = ""
}
parentPath = filepath.FromSlash(parentRel) parentPath = filepath.FromSlash(parentRel)
newPath = filepath.Join(repo.Path, parentPath, newName) newPath = filepath.Join(repo.Path, parentPath, newName)
newRelPath = filepath.ToSlash(filepath.Join(parentRel, newName)) newRelPath = filepath.ToSlash(filepath.Join(parentRel, newName))
_, err = api.store(r).GetRPMRepoDir(repo.ID, newRelPath) _, err = api.store(r).GetRPMRepoDir(repo.ID, newRelPath)
if err == nil { if err == nil {
targetConfigExists = true targetConfigExists = true
@@ -4754,20 +4940,14 @@ func (api *API) RepoRPMRenameSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
err = os.Rename(fullPath, newPath) err = os.Rename(fullPath, newPath)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
}
} }
err = api.store(r).MoveRPMRepoDir(r.Context(), repo.ID, relPathClean, newRelPath) if isRepoDir {
if err != nil { dirConfig = models.RPMRepoDir{
_ = os.Rename(newPath, fullPath)
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
}
if isRepoDir {
dirConfig = models.RPMRepoDir{
RepoID: repo.ID, RepoID: repo.ID,
Path: newRelPath, Path: newRelPath,
Mode: newMode, Mode: newMode,
@@ -4788,14 +4968,24 @@ func (api *API) RepoRPMRenameSubdir(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error()) WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return return
} }
dirConfig.SyncEnabled = true dirConfig.SyncEnabled = true
} }
err = api.store(r).UpsertRPMRepoDir(dirConfig)
if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
} }
if renamed || isRepoDir {
ctx, cancel = context.WithTimeout(context.Background(), db.IndependentDBOperationTimeout)
defer cancel()
txStore, err = api.Store.BeginImmediateStore(ctx)
if err == nil && renamed { err = txStore.MoveRPMRepoDir(ctx, repo.ID, relPathClean, newRelPath) }
if err == nil && isRepoDir { err = txStore.UpsertRPMRepoDir(dirConfig) }
if err == nil { err = txStore.Commit() }
if err != nil {
if txStore != nil { _ = txStore.Rollback() }
if renamed { _ = os.Rename(newPath, fullPath) }
WriteJSONWithErrorReason(w, r, http.StatusInternalServerError, err.Error())
return
}
} }
repodataPath = filepath.Join(filepath.Join(repo.Path, parentPath), "repodata") repodataPath = filepath.Join(filepath.Join(repo.Path, parentPath), "repodata")
_, err = os.Stat(repodataPath) _, err = os.Stat(repodataPath)
if err == nil && api.RpmMeta != nil { if err == nil && api.RpmMeta != nil {
@@ -4817,6 +5007,8 @@ func (api *API) RepoRPMDeleteFile(w http.ResponseWriter, r *http.Request, params
var parentPath string var parentPath string
var repodataPath string var repodataPath string
var lower string var lower string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -4829,6 +5021,8 @@ func (api *API) RepoRPMDeleteFile(w http.ResponseWriter, r *http.Request, params
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
if relPath == "" { if relPath == "" {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "path required")
@@ -5053,6 +5247,7 @@ func (api *API) RepoRPMUpload(w http.ResponseWriter, r *http.Request, params map
var info os.FileInfo var info os.FileInfo
var oldTemp string var oldTemp string
var ts string var ts string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
@@ -5066,6 +5261,8 @@ func (api *API) RepoRPMUpload(w http.ResponseWriter, r *http.Request, params map
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not rpm")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
relPath = strings.TrimSpace(r.URL.Query().Get("path")) relPath = strings.TrimSpace(r.URL.Query().Get("path"))
overwriteParam = strings.TrimSpace(r.URL.Query().Get("overwrite")) overwriteParam = strings.TrimSpace(r.URL.Query().Get("overwrite"))
overwriteParam = strings.ToLower(overwriteParam) overwriteParam = strings.ToLower(overwriteParam)
@@ -5501,6 +5698,8 @@ func (api *API) RepoDockerDeleteTag(w http.ResponseWriter, r *http.Request, para
var image string var image string
var tag string var tag string
var imagePath string var imagePath string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -5513,6 +5712,8 @@ func (api *API) RepoDockerDeleteTag(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
image = strings.TrimSpace(r.URL.Query().Get("image")) image = strings.TrimSpace(r.URL.Query().Get("image"))
tag = strings.TrimSpace(r.URL.Query().Get("tag")) tag = strings.TrimSpace(r.URL.Query().Get("tag"))
if tag == "" { if tag == "" {
@@ -5532,6 +5733,8 @@ func (api *API) RepoDockerDeleteImage(w http.ResponseWriter, r *http.Request, pa
var repo models.Repo var repo models.Repo
var err error var err error
var image string var image string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -5544,6 +5747,8 @@ func (api *API) RepoDockerDeleteImage(w http.ResponseWriter, r *http.Request, pa
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
image = strings.TrimSpace(r.URL.Query().Get("image")) image = strings.TrimSpace(r.URL.Query().Get("image"))
err = docker.DeleteImage(repo.Path, image) err = docker.DeleteImage(repo.Path, image)
if err != nil { if err != nil {
@@ -5561,6 +5766,8 @@ func (api *API) RepoDockerRenameTag(w http.ResponseWriter, r *http.Request, para
var from string var from string
var to string var to string
var imagePath string var imagePath string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -5573,6 +5780,8 @@ func (api *API) RepoDockerRenameTag(w http.ResponseWriter, r *http.Request, para
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
@@ -5604,6 +5813,8 @@ func (api *API) RepoDockerRenameImage(w http.ResponseWriter, r *http.Request, pa
var req repoDockerRenameImageRequest var req repoDockerRenameImageRequest
var from string var from string
var to string var to string
var unlock func()
repo, err = api.getRepoResolved(api.store(r), params["id"]) repo, err = api.getRepoResolved(api.store(r), params["id"])
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found") WriteJSONWithErrorReason(w, r, http.StatusNotFound, "repo not found")
@@ -5616,6 +5827,8 @@ func (api *API) RepoDockerRenameImage(w http.ResponseWriter, r *http.Request, pa
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "repo is not docker")
return return
} }
unlock = api.lockRepo(repo.ID)
defer unlock()
err = DecodeJSON(r, &req) err = DecodeJSON(r, &req)
if err != nil { if err != nil {
WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json") WriteJSONWithErrorReason(w, r, http.StatusBadRequest, "invalid json")
+2 -3
View File
@@ -421,9 +421,8 @@ func (api *API) DeleteBlockAttachment(w http.ResponseWriter, r *http.Request, pa
var item models.BlockAttachment var item models.BlockAttachment
var err error var err error
if !api.requireBoardRole(w, r, params["boardId"], models.RoleWriter) { if !api.requireBoardRole(w, r, params["boardId"], models.RoleWriter) { return }
return
}
item, err = api.store(r).DeleteBlockAttachment(r.Context(), params["boardId"], params["blockId"], params["attachmentId"]) item, err = api.store(r).DeleteBlockAttachment(r.Context(), params["boardId"], params["blockId"], params["attachmentId"])
if err != nil { if err != nil {
if errors.Is(err, db.ErrBlockAttachmentNotFound) { if errors.Is(err, db.ErrBlockAttachmentNotFound) {
+112
View File
@@ -0,0 +1,112 @@
package repolock
import "sort"
import "strings"
import "sync"
type Manager struct {
locks sync.Map
}
func NewManager() *Manager {
return &Manager{}
}
func (m *Manager) Lock(repoID string) func() {
var key string
var value any
var actual any
var lock *sync.Mutex
var ok bool
if m == nil { return func() {} }
key = strings.TrimSpace(repoID)
if key == "" { key = "-" }
value = &sync.Mutex{}
actual, _ = m.locks.LoadOrStore(key, value)
lock, ok = actual.(*sync.Mutex)
if !ok || lock == nil { return func() {} }
lock.Lock()
return func() {
lock.Unlock()
}
}
func (m *Manager) TryLock(repoID string) (func(), bool) {
var key string
var value any
var actual any
var lock *sync.Mutex
var ok bool
if m == nil { return func() {}, true }
key = strings.TrimSpace(repoID)
if key == "" { key = "-" }
value = &sync.Mutex{}
actual, _ = m.locks.LoadOrStore(key, value)
lock, ok = actual.(*sync.Mutex)
if !ok || lock == nil { return func() {}, true }
if !lock.TryLock() { return nil, false }
return func() {
lock.Unlock()
}, true
}
func (m *Manager) LockMany(repoIDs []string) func() {
var ids []string
var seen map[string]bool
var unlocks []func()
var id string
var i int
seen = make(map[string]bool)
for i = 0; i < len(repoIDs); i++ {
id = strings.TrimSpace(repoIDs[i])
if id == "" || seen[id] { continue }
seen[id] = true
ids = append(ids, id)
}
sort.Strings(ids)
for i = 0; i < len(ids); i++ {
unlocks = append(unlocks, m.Lock(ids[i]))
}
return func() {
for i = len(unlocks) - 1; i >= 0; i-- {
if unlocks[i] != nil { unlocks[i]() }
}
}
}
func (m *Manager) TryLockMany(repoIDs []string) (func(), bool) {
var ids []string
var seen map[string]bool
var unlocks []func()
var unlock func()
var ok bool
var id string
var i int
seen = make(map[string]bool)
for i = 0; i < len(repoIDs); i++ {
id = strings.TrimSpace(repoIDs[i])
if id == "" || seen[id] { continue }
seen[id] = true
ids = append(ids, id)
}
sort.Strings(ids)
for i = 0; i < len(ids); i++ {
unlock, ok = m.TryLock(ids[i])
if !ok {
for i = len(unlocks) - 1; i >= 0; i-- {
if unlocks[i] != nil { unlocks[i]() }
}
return nil, false
}
unlocks = append(unlocks, unlock)
}
return func() {
for i = len(unlocks) - 1; i >= 0; i-- {
if unlocks[i] != nil { unlocks[i]() }
}
}, true
}
+55 -44
View File
@@ -32,6 +32,7 @@ import codit_http "codit/internal/http"
import "codit/internal/middleware" import "codit/internal/middleware"
import "codit/internal/models" import "codit/internal/models"
import "codit/internal/pkiutil" import "codit/internal/pkiutil"
import "codit/internal/repolock"
import "codit/internal/rpm" import "codit/internal/rpm"
import "codit/internal/storage" import "codit/internal/storage"
import "codit/internal/util" import "codit/internal/util"
@@ -52,11 +53,13 @@ type server_http_log_writer struct {
type gitPathRewriteHandler struct { type gitPathRewriteHandler struct {
next http.Handler next http.Handler
store *db.Store store *db.Store
locks *repolock.Manager
} }
type gitIDPathRewriteHandler struct { type gitIDPathRewriteHandler struct {
next http.Handler next http.Handler
store *db.Store store *db.Store
locks *repolock.Manager
} }
type rpmPathRewriteHandler struct { type rpmPathRewriteHandler struct {
@@ -157,6 +160,7 @@ type Server struct {
rpm_meta *rpm.MetaManager rpm_meta *rpm.MetaManager
rpm_mirror *rpm.MirrorManager rpm_mirror *rpm.MirrorManager
upload_store *storage.FileStore upload_store *storage.FileStore
repo_locks *repolock.Manager
docker_server http.Handler docker_server http.Handler
git_server http.Handler git_server http.Handler
@@ -188,6 +192,8 @@ func (h *gitPathRewriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
var repoStorageID int64 var repoStorageID int64
var err error var err error
var newPath string var newPath string
var unlock func()
var locked bool
currentStore = requestStore(r, h.store) currentStore = requestStore(r, h.store)
path = strings.TrimPrefix(r.URL.Path, "/") path = strings.TrimPrefix(r.URL.Path, "/")
@@ -233,6 +239,14 @@ func (h *gitPathRewriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request
RepoName: repo.Name, RepoName: repo.Name,
}) })
r.URL.Path = newPath r.URL.Path = newPath
if classifyServiceOperation(serviceGit, r) == opWrite {
unlock, locked = h.locks.TryLock(repo.ID)
if !locked {
http.Error(w, "repository is busy", http.StatusConflict)
return
}
defer unlock()
}
h.next.ServeHTTP(w, r) h.next.ServeHTTP(w, r)
} }
@@ -250,6 +264,8 @@ func (h *gitIDPathRewriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Reque
var err error var err error
var newPath string var newPath string
var repoID string var repoID string
var unlock func()
var locked bool
currentStore = requestStore(r, h.store) currentStore = requestStore(r, h.store)
path = strings.TrimPrefix(r.URL.Path, "/") path = strings.TrimPrefix(r.URL.Path, "/")
if path == "" { if path == "" {
@@ -289,6 +305,14 @@ func (h *gitIDPathRewriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Reque
RepoName: repo.Name, RepoName: repo.Name,
}) })
r.URL.Path = newPath r.URL.Path = newPath
if classifyServiceOperation(serviceGit, r) == opWrite {
unlock, locked = h.locks.TryLock(repo.ID)
if !locked {
http.Error(w, "repository is busy", http.StatusConflict)
return
}
defer unlock()
}
h.next.ServeHTTP(w, r) h.next.ServeHTTP(w, r)
} }
@@ -589,6 +613,7 @@ func newServerWithInternalConfig(cfg *codit_config.Config, logger Logger, identi
app.rpm_meta = rpm.NewMetaManager() app.rpm_meta = rpm.NewMetaManager()
app.rpm_mirror = rpm.NewMirrorManager(app.store, logger, app.rpm_meta) app.rpm_mirror = rpm.NewMirrorManager(app.store, logger, app.rpm_meta)
app.upload_store = &storage.FileStore{BaseDir: app.uploads_base_dir} app.upload_store = &storage.FileStore{BaseDir: app.uploads_base_dir}
app.repo_locks = repolock.NewManager()
for _, dir = range []string{app.docker_base_dir, app.git_base_dir, app.rpm_base_dir, app.uploads_base_dir} { for _, dir = range []string{app.docker_base_dir, app.git_base_dir, app.rpm_base_dir, app.uploads_base_dir} {
err = os.MkdirAll(dir, 0o755) err = os.MkdirAll(dir, 0o755)
@@ -598,7 +623,7 @@ func newServerWithInternalConfig(cfg *codit_config.Config, logger Logger, identi
} }
} }
app.docker_server = docker.NewHTTPServer(app.store, app.docker_base_dir, nil, logger, cfg.DockerHTTPPrefix) app.docker_server = docker.NewHTTPServer(app.store, app.docker_base_dir, nil, logger, cfg.DockerHTTPPrefix, app.repo_locks)
app.git_server, err = git.NewHTTPServer(app.git_base_dir, nil, logger) app.git_server, err = git.NewHTTPServer(app.git_base_dir, nil, logger)
if err != nil { if err != nil {
@@ -733,6 +758,7 @@ func (app *Server) initHandlers() (*handlers.API, *http.ServeMux, error) {
DockerBase: app.docker_base_dir, DockerBase: app.docker_base_dir,
Uploads: *app.upload_store, Uploads: *app.upload_store,
Logger: logger, Logger: logger,
RepoLocks: app.repo_locks,
}) })
router = codit_http.NewRouter() router = codit_http.NewRouter()
@@ -754,8 +780,8 @@ func (app *Server) initHandlers() (*handlers.API, *http.ServeMux, error) {
router.Handle("GET", "/api/auth/oidc/:id/callback", api.OIDCCallbackWithProvider) router.Handle("GET", "/api/auth/oidc/:id/callback", api.OIDCCallbackWithProvider)
router.Handle("GET", "/api/me", txRead(api.Me)) router.Handle("GET", "/api/me", txRead(api.Me))
router.Handle("PATCH", "/api/me", txWrite(api.UpdateMe)) router.Handle("PATCH", "/api/me", txWrite(api.UpdateMe))
router.Handle("POST", "/api/me/avatar", txWrite(api.UploadMyAvatar)) router.Handle("POST", "/api/me/avatar", api.UploadMyAvatar)
router.Handle("DELETE", "/api/me/avatar", txWrite(api.DeleteMyAvatar)) router.Handle("DELETE", "/api/me/avatar", api.DeleteMyAvatar)
router.Handle("GET", "/api/me/totp", txRead(api.GetMyTOTP)) router.Handle("GET", "/api/me/totp", txRead(api.GetMyTOTP))
router.Handle("POST", "/api/me/totp/setup", txWrite(api.SetupMyTOTP)) router.Handle("POST", "/api/me/totp/setup", txWrite(api.SetupMyTOTP))
router.Handle("POST", "/api/me/totp/enable", txWrite(api.EnableMyTOTP)) router.Handle("POST", "/api/me/totp/enable", txWrite(api.EnableMyTOTP))
@@ -987,7 +1013,7 @@ func (app *Server) initHandlers() (*handlers.API, *http.ServeMux, error) {
router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/attachments", txRead(api.ListBlockAttachments)) router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/attachments", txRead(api.ListBlockAttachments))
router.Handle("POST", "/api/boards/:boardId/blocks/:blockId/attachments", api.CreateBlockAttachment) router.Handle("POST", "/api/boards/:boardId/blocks/:blockId/attachments", api.CreateBlockAttachment)
router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/attachments/:attachmentId/download", api.DownloadBlockAttachment) router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/attachments/:attachmentId/download", api.DownloadBlockAttachment)
router.Handle("DELETE", "/api/boards/:boardId/blocks/:blockId/attachments/:attachmentId", txWrite(api.DeleteBlockAttachment)) router.Handle("DELETE", "/api/boards/:boardId/blocks/:blockId/attachments/:attachmentId", api.DeleteBlockAttachment)
router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/properties", txRead(api.GetBlockProperties)) router.Handle("GET", "/api/boards/:boardId/blocks/:blockId/properties", txRead(api.GetBlockProperties))
router.Handle("PUT", "/api/boards/:boardId/blocks/:blockId/properties", txWrite(api.UpsertBlockProperties)) router.Handle("PUT", "/api/boards/:boardId/blocks/:blockId/properties", txWrite(api.UpsertBlockProperties))
router.Handle("GET", "/api/boards/:boardId/members", txRead(api.ListBoardMembers)) router.Handle("GET", "/api/boards/:boardId/members", txRead(api.ListBoardMembers))
@@ -1023,19 +1049,19 @@ func (app *Server) initHandlers() (*handlers.API, *http.ServeMux, error) {
router.Handle("GET", "/api/repos/:id/stats", api.RepoStats) router.Handle("GET", "/api/repos/:id/stats", api.RepoStats)
router.Handle("GET", "/api/repos/:id/rpm/packages", txRead(api.RepoRPMPackages)) router.Handle("GET", "/api/repos/:id/rpm/packages", txRead(api.RepoRPMPackages))
router.Handle("GET", "/api/repos/:id/rpm/package", txRead(api.RepoRPMPackage)) router.Handle("GET", "/api/repos/:id/rpm/package", txRead(api.RepoRPMPackage))
router.Handle("POST", "/api/repos/:id/rpm/subdirs", txWrite(api.RepoRPMCreateSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdirs", api.RepoRPMCreateSubdir)
router.Handle("GET", "/api/repos/:id/rpm/subdir", txRead(api.RepoRPMGetSubdir)) router.Handle("GET", "/api/repos/:id/rpm/subdir", txRead(api.RepoRPMGetSubdir))
router.Handle("POST", "/api/repos/:id/rpm/subdir/update", txWrite(api.RepoRPMRenameSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdir/update", api.RepoRPMRenameSubdir)
router.Handle("POST", "/api/repos/:id/rpm/subdir/rename", txWrite(api.RepoRPMRenameSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdir/rename", api.RepoRPMRenameSubdir)
router.Handle("POST", "/api/repos/:id/rpm/subdir/sync", txWrite(api.RepoRPMSyncSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdir/sync", api.RepoRPMSyncSubdir)
router.Handle("POST", "/api/repos/:id/rpm/subdir/suspend", txWrite(api.RepoRPMSuspendSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdir/suspend", api.RepoRPMSuspendSubdir)
router.Handle("POST", "/api/repos/:id/rpm/subdir/resume", txWrite(api.RepoRPMResumeSubdir)) router.Handle("POST", "/api/repos/:id/rpm/subdir/resume", api.RepoRPMResumeSubdir)
router.Handle("POST", "/api/repos/:id/rpm/subdir/rebuild-metadata", api.RepoRPMRebuildSubdirMetadata) router.Handle("POST", "/api/repos/:id/rpm/subdir/rebuild-metadata", api.RepoRPMRebuildSubdirMetadata)
router.Handle("POST", "/api/repos/:id/rpm/subdir/cancel", txWrite(api.RepoRPMCancelSubdirSync)) router.Handle("POST", "/api/repos/:id/rpm/subdir/cancel", api.RepoRPMCancelSubdirSync)
router.Handle("GET", "/api/repos/:id/rpm/subdir/runs", txRead(api.RepoRPMMirrorRuns)) router.Handle("GET", "/api/repos/:id/rpm/subdir/runs", txRead(api.RepoRPMMirrorRuns))
router.Handle("DELETE", "/api/repos/:id/rpm/subdir/runs", txWrite(api.RepoRPMClearMirrorRuns)) router.Handle("DELETE", "/api/repos/:id/rpm/subdir/runs", txWrite(api.RepoRPMClearMirrorRuns))
router.Handle("DELETE", "/api/repos/:id/rpm/subdir", txWrite(api.RepoRPMDeleteSubdir)) router.Handle("DELETE", "/api/repos/:id/rpm/subdir", api.RepoRPMDeleteSubdir)
router.Handle("DELETE", "/api/repos/:id/rpm/file", txWrite(api.RepoRPMDeleteFile)) router.Handle("DELETE", "/api/repos/:id/rpm/file", api.RepoRPMDeleteFile)
router.Handle("GET", "/api/repos/:id/rpm/file", api.RepoRPMFile) router.Handle("GET", "/api/repos/:id/rpm/file", api.RepoRPMFile)
router.Handle("GET", "/api/repos/:id/rpm/tree", txRead(api.RepoRPMTree)) router.Handle("GET", "/api/repos/:id/rpm/tree", txRead(api.RepoRPMTree))
router.Handle("POST", "/api/repos/:id/rpm/upload", api.RepoRPMUpload) router.Handle("POST", "/api/repos/:id/rpm/upload", api.RepoRPMUpload)
@@ -1086,32 +1112,27 @@ func (app *Server) initHandlers() (*handlers.API, *http.ServeMux, error) {
mux = http.NewServeMux() mux = http.NewServeMux()
gitRewrite = &gitPathRewriteHandler{next: app.git_server, store: store} gitRewrite = &gitPathRewriteHandler{next: app.git_server, store: store, locks: app.repo_locks}
gitHandler = withServiceAuth(serviceGit, cfg.GitHTTPPrefix, http.StripPrefix(cfg.GitHTTPPrefix, gitRewrite), auth_user, store, logger) gitHandler = withServiceAuth(serviceGit, cfg.GitHTTPPrefix, http.StripPrefix(cfg.GitHTTPPrefix, gitRewrite), auth_user, store, logger)
gitHandler = middleware.WithStoreTransactionUnbuffered(store, gitHandler)
gitHandler = middleware.WithRequestStore(store, gitHandler) gitHandler = middleware.WithRequestStore(store, gitHandler)
mux.Handle(cfg.GitHTTPPrefix+"/", gitHandler) mux.Handle(cfg.GitHTTPPrefix+"/", gitHandler)
gitIDRewrite = &gitIDPathRewriteHandler{next: app.git_server, store: store} gitIDRewrite = &gitIDPathRewriteHandler{next: app.git_server, store: store, locks: app.repo_locks}
gitIDHandler = withServiceAuth(serviceGit, cfg.GitHTTPPrefix+"-id", http.StripPrefix(cfg.GitHTTPPrefix+"-id", gitIDRewrite), auth_user, store, logger) gitIDHandler = withServiceAuth(serviceGit, cfg.GitHTTPPrefix+"-id", http.StripPrefix(cfg.GitHTTPPrefix+"-id", gitIDRewrite), auth_user, store, logger)
gitIDHandler = middleware.WithStoreTransactionUnbuffered(store, gitIDHandler)
gitIDHandler = middleware.WithRequestStore(store, gitIDHandler) gitIDHandler = middleware.WithRequestStore(store, gitIDHandler)
mux.Handle(cfg.GitHTTPPrefix+"-id/", gitIDHandler) mux.Handle(cfg.GitHTTPPrefix+"-id/", gitIDHandler)
rpmRewrite = &rpmPathRewriteHandler{next: app.rpm_server, store: store} rpmRewrite = &rpmPathRewriteHandler{next: app.rpm_server, store: store}
rpmHandler = withServiceAuth(serviceRPM, cfg.RPMHTTPPrefix, http.StripPrefix(cfg.RPMHTTPPrefix, rpmRewrite), auth_user, store, logger) rpmHandler = withServiceAuth(serviceRPM, cfg.RPMHTTPPrefix, http.StripPrefix(cfg.RPMHTTPPrefix, rpmRewrite), auth_user, store, logger)
rpmHandler = middleware.WithStoreTransactionUnbuffered(store, rpmHandler)
rpmHandler = middleware.WithRequestStore(store, rpmHandler) rpmHandler = middleware.WithRequestStore(store, rpmHandler)
mux.Handle(cfg.RPMHTTPPrefix+"/", rpmHandler) mux.Handle(cfg.RPMHTTPPrefix+"/", rpmHandler)
rpmIDRewrite = &rpmIDPathRewriteHandler{next: app.rpm_server, store: store} rpmIDRewrite = &rpmIDPathRewriteHandler{next: app.rpm_server, store: store}
rpmIDHandler = withServiceAuth(serviceRPM, cfg.RPMHTTPPrefix+"-id", http.StripPrefix(cfg.RPMHTTPPrefix+"-id", rpmIDRewrite), auth_user, store, logger) rpmIDHandler = withServiceAuth(serviceRPM, cfg.RPMHTTPPrefix+"-id", http.StripPrefix(cfg.RPMHTTPPrefix+"-id", rpmIDRewrite), auth_user, store, logger)
rpmIDHandler = middleware.WithStoreTransactionUnbuffered(store, rpmIDHandler)
rpmIDHandler = middleware.WithRequestStore(store, rpmIDHandler) rpmIDHandler = middleware.WithRequestStore(store, rpmIDHandler)
mux.Handle(cfg.RPMHTTPPrefix+"-id/", rpmIDHandler) mux.Handle(cfg.RPMHTTPPrefix+"-id/", rpmIDHandler)
dockerHandler = withServiceAuth(serviceV2, cfg.DockerHTTPPrefix, app.docker_server, auth_user, store, logger) dockerHandler = withServiceAuth(serviceV2, cfg.DockerHTTPPrefix, app.docker_server, auth_user, store, logger)
dockerHandler = middleware.WithStoreTransactionUnbuffered(store, dockerHandler)
dockerHandler = middleware.WithRequestStore(store, dockerHandler) dockerHandler = middleware.WithRequestStore(store, dockerHandler)
mux.Handle(cfg.DockerHTTPPrefix, dockerHandler) mux.Handle(cfg.DockerHTTPPrefix, dockerHandler)
mux.Handle(cfg.DockerHTTPPrefix+"/", dockerHandler) mux.Handle(cfg.DockerHTTPPrefix+"/", dockerHandler)
@@ -1739,36 +1760,26 @@ func withAPIAuth(next http.Handler, auth func(store *db.Store, username string,
func classifyServiceOperation(service serviceKind, r *http.Request) opKind { func classifyServiceOperation(service serviceKind, r *http.Request) opKind {
var serviceName string var serviceName string
var path string
var method string
serviceName = "" serviceName = ""
if r != nil { if r != nil {
serviceName = r.URL.Query().Get("service") serviceName = r.URL.Query().Get("service")
path = r.URL.Path
method = r.Method
} }
if service == serviceGit { if service == serviceGit {
if strings.Contains(r.URL.Path, "git-receive-pack") || serviceName == "git-receive-pack" { if strings.Contains(path, "git-receive-pack") || serviceName == "git-receive-pack" { return opWrite }
return opWrite } else if service == serviceAPI {
} if method != http.MethodGet && method != http.MethodHead && method != http.MethodOptions { return opWrite }
return opRead } else if service == serviceRPM {
} if method != http.MethodGet && method != http.MethodHead && method != http.MethodOptions { return opWrite }
if service == serviceAPI { } else if service == serviceV2 {
if r.Method == http.MethodGet || r.Method == http.MethodHead || r.Method == http.MethodOptions { if method == http.MethodPost || method == http.MethodPut || method == http.MethodPatch || method == http.MethodDelete { return opWrite }
return opRead
}
return opWrite
}
if service == serviceRPM {
if r.Method == http.MethodGet || r.Method == http.MethodHead || r.Method == http.MethodOptions {
return opRead
}
return opWrite
}
if service == serviceV2 {
if r.Method == http.MethodGet || r.Method == http.MethodHead {
return opRead
}
if r.Method == http.MethodPost || r.Method == http.MethodPut || r.Method == http.MethodPatch || r.Method == http.MethodDelete {
return opWrite
}
} }
return opRead return opRead
} }