DevFW-CICD/garm
15
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Add write lock for sqlite3

Browse source 
Signed-off-by: Gabriel Adrian Samfira <gsamfira@cloudbasesolutions.com>
This commit is contained in:
Gabriel Adrian Samfira 2025-05-26 19:12:59 +00:00
parent 14b253b9ce
commit ef3402bf17
10 changed files with 105 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

6
database/sql/controller.go
View file

@ -63,6 +63,9 @@ func (s *sqlDatabase) ControllerInfo() (params.ControllerInfo, error) {
}
func (s *sqlDatabase) InitController() (params.ControllerInfo, error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
if _, err := s.ControllerInfo(); err == nil {
return params.ControllerInfo{}, runnerErrors.NewConflictError("controller already initialized")
}
@ -88,6 +91,9 @@ func (s *sqlDatabase) InitController() (params.ControllerInfo, error) {
}
func (s *sqlDatabase) UpdateController(info params.UpdateControllerParams) (paramInfo params.ControllerInfo, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.ControllerEntityType, common.UpdateOperation, paramInfo)

9
database/sql/enterprise.go
View file

@ -29,6 +29,9 @@ import (
)
func (s *sqlDatabase) CreateEnterprise(ctx context.Context, name, credentialsName, webhookSecret string, poolBalancerType params.PoolBalancerType) (paramEnt params.Enterprise, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
if webhookSecret == "" {
return params.Enterprise{}, errors.New("creating enterprise: missing secret")
}
@ -132,6 +135,9 @@ func (s *sqlDatabase) ListEnterprises(_ context.Context) ([]params.Enterprise, e
}
func (s *sqlDatabase) DeleteEnterprise(ctx context.Context, enterpriseID string) error {
s.writeMux.Lock()
defer s.writeMux.Unlock()
enterprise, err := s.getEnterpriseByID(ctx, s.conn, enterpriseID, "Endpoint", "Credentials", "Credentials.Endpoint")
if err != nil {
return errors.Wrap(err, "fetching enterprise")
@ -157,6 +163,9 @@ func (s *sqlDatabase) DeleteEnterprise(ctx context.Context, enterpriseID string)
}
func (s *sqlDatabase) UpdateEnterprise(ctx context.Context, enterpriseID string, param params.UpdateEntityParams) (newParams params.Enterprise, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.EnterpriseEntityType, common.UpdateOperation, newParams)

18
database/sql/github.go
View file

@ -111,6 +111,9 @@ func getUIDFromContext(ctx context.Context) (uuid.UUID, error) {
}
func (s *sqlDatabase) CreateGithubEndpoint(_ context.Context, param params.CreateGithubEndpointParams) (ghEndpoint params.GithubEndpoint, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.GithubEndpointEntityType, common.CreateOperation, ghEndpoint)
@ -164,6 +167,9 @@ func (s *sqlDatabase) ListGithubEndpoints(_ context.Context) ([]params.GithubEnd
}
func (s *sqlDatabase) UpdateGithubEndpoint(_ context.Context, name string, param params.UpdateGithubEndpointParams) (ghEndpoint params.GithubEndpoint, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.GithubEndpointEntityType, common.UpdateOperation, ghEndpoint)
@ -229,6 +235,9 @@ func (s *sqlDatabase) GetGithubEndpoint(_ context.Context, name string) (params.
}
func (s *sqlDatabase) DeleteGithubEndpoint(_ context.Context, name string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.GithubEndpointEntityType, common.DeleteOperation, params.GithubEndpoint{Name: name})
@ -287,6 +296,9 @@ func (s *sqlDatabase) DeleteGithubEndpoint(_ context.Context, name string) (err
}
func (s *sqlDatabase) CreateGithubCredentials(ctx context.Context, param params.CreateGithubCredentialsParams) (ghCreds params.GithubCredentials, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
userID, err := getUIDFromContext(ctx)
if err != nil {
return params.GithubCredentials{}, errors.Wrap(err, "creating github credentials")
@ -450,6 +462,9 @@ func (s *sqlDatabase) ListGithubCredentials(ctx context.Context) ([]params.Githu
}
func (s *sqlDatabase) UpdateGithubCredentials(ctx context.Context, id uint, param params.UpdateGithubCredentialsParams) (ghCreds params.GithubCredentials, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.GithubCredentialsEntityType, common.UpdateOperation, ghCreds)
@ -529,6 +544,9 @@ func (s *sqlDatabase) UpdateGithubCredentials(ctx context.Context, id uint, para
}
func (s *sqlDatabase) DeleteGithubCredentials(ctx context.Context, id uint) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
var name string
defer func() {
if err == nil {

12
database/sql/instances.go
View file

@ -31,6 +31,9 @@ import (
)
func (s *sqlDatabase) CreateInstance(_ context.Context, poolID string, param params.CreateInstanceParams) (instance params.Instance, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
pool, err := s.getPoolByID(s.conn, poolID)
if err != nil {
return params.Instance{}, errors.Wrap(err, "fetching pool")
@ -143,6 +146,9 @@ func (s *sqlDatabase) GetInstanceByName(ctx context.Context, instanceName string
}
func (s *sqlDatabase) DeleteInstance(_ context.Context, poolID string, instanceName string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
instance, err := s.getPoolInstanceByName(poolID, instanceName)
if err != nil {
return errors.Wrap(err, "deleting instance")
@ -176,6 +182,9 @@ func (s *sqlDatabase) DeleteInstance(_ context.Context, poolID string, instanceN
}
func (s *sqlDatabase) AddInstanceEvent(ctx context.Context, instanceName string, event params.EventType, eventLevel params.EventLevel, statusMessage string) error {
s.writeMux.Lock()
defer s.writeMux.Unlock()
instance, err := s.getInstanceByName(ctx, instanceName)
if err != nil {
return errors.Wrap(err, "updating instance")
@ -194,6 +203,9 @@ func (s *sqlDatabase) AddInstanceEvent(ctx context.Context, instanceName string,
}
func (s *sqlDatabase) UpdateInstance(ctx context.Context, instanceName string, param params.UpdateInstanceParams) (params.Instance, error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
instance, err := s.getInstanceByName(ctx, instanceName)
if err != nil {
return params.Instance{}, errors.Wrap(err, "updating instance")

18
database/sql/jobs.go
View file

@ -95,6 +95,9 @@ func (s *sqlDatabase) paramsJobToWorkflowJob(ctx context.Context, job params.Job
}
func (s *sqlDatabase) DeleteJob(_ context.Context, jobID int64) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
if notifyErr := s.sendNotify(common.JobEntityType, common.DeleteOperation, params.Job{ID: jobID}); notifyErr != nil {
@ -113,6 +116,9 @@ func (s *sqlDatabase) DeleteJob(_ context.Context, jobID int64) (err error) {
}
func (s *sqlDatabase) LockJob(_ context.Context, jobID int64, entityID string) error {
s.writeMux.Lock()
defer s.writeMux.Unlock()
entityUUID, err := uuid.Parse(entityID)
if err != nil {
return errors.Wrap(err, "parsing entity id")
@ -152,6 +158,9 @@ func (s *sqlDatabase) LockJob(_ context.Context, jobID int64, entityID string) e
}
func (s *sqlDatabase) BreakLockJobIsQueued(_ context.Context, jobID int64) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
var workflowJob WorkflowJob
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("id = ? and status = ?", jobID, params.JobStatusQueued).First(&workflowJob)
@ -180,6 +189,9 @@ func (s *sqlDatabase) BreakLockJobIsQueued(_ context.Context, jobID int64) (err
}
func (s *sqlDatabase) UnlockJob(_ context.Context, jobID int64, entityID string) error {
s.writeMux.Lock()
defer s.writeMux.Unlock()
var workflowJob WorkflowJob
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", jobID).First(&workflowJob)
@ -213,6 +225,9 @@ func (s *sqlDatabase) UnlockJob(_ context.Context, jobID int64, entityID string)
}
func (s *sqlDatabase) CreateOrUpdateJob(ctx context.Context, job params.Job) (params.Job, error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
var workflowJob WorkflowJob
var err error
q := s.conn.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Instance").Where("id = ?", job.ID).First(&workflowJob)
@ -381,6 +396,9 @@ func (s *sqlDatabase) GetJobByID(_ context.Context, jobID int64) (params.Job, er
// DeleteCompletedJobs deletes all completed jobs.
func (s *sqlDatabase) DeleteCompletedJobs(_ context.Context) error {
s.writeMux.Lock()
defer s.writeMux.Unlock()
query := s.conn.Model(&WorkflowJob{}).Where("status = ?", params.JobStatusCompleted)
if err := query.Unscoped().Delete(&WorkflowJob{}); err.Error != nil {

9
database/sql/organizations.go
View file

@ -30,6 +30,9 @@ import (
)
func (s *sqlDatabase) CreateOrganization(ctx context.Context, name, credentialsName, webhookSecret string, poolBalancerType params.PoolBalancerType) (org params.Organization, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
if webhookSecret == "" {
return params.Organization{}, errors.New("creating org: missing secret")
}
@ -123,6 +126,9 @@ func (s *sqlDatabase) ListOrganizations(_ context.Context) ([]params.Organizatio
}
func (s *sqlDatabase) DeleteOrganization(ctx context.Context, orgID string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
org, err := s.getOrgByID(ctx, s.conn, orgID, "Endpoint", "Credentials", "Credentials.Endpoint")
if err != nil {
return errors.Wrap(err, "fetching org")
@ -148,6 +154,9 @@ func (s *sqlDatabase) DeleteOrganization(ctx context.Context, orgID string) (err
}
func (s *sqlDatabase) UpdateOrganization(ctx context.Context, orgID string, param params.UpdateEntityParams) (paramOrg params.Organization, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.OrganizationEntityType, common.UpdateOperation, paramOrg)

12
database/sql/pools.go
View file

@ -68,6 +68,9 @@ func (s *sqlDatabase) GetPoolByID(_ context.Context, poolID string) (params.Pool
}
func (s *sqlDatabase) DeletePoolByID(_ context.Context, poolID string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
pool, err := s.getPoolByID(s.conn, poolID)
if err != nil {
return errors.Wrap(err, "fetching pool by ID")
@ -255,6 +258,9 @@ func (s *sqlDatabase) FindPoolsMatchingAllTags(_ context.Context, entityType par
}
func (s *sqlDatabase) CreateEntityPool(_ context.Context, entity params.GithubEntity, param params.CreatePoolParams) (pool params.Pool, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
if len(param.Tags) == 0 {
return params.Pool{}, runnerErrors.NewBadRequestError("no tags specified")
}
@ -343,6 +349,9 @@ func (s *sqlDatabase) GetEntityPool(_ context.Context, entity params.GithubEntit
}
func (s *sqlDatabase) DeleteEntityPool(_ context.Context, entity params.GithubEntity, poolID string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
entityID, err := uuid.Parse(entity.ID)
if err != nil {
return errors.Wrap(runnerErrors.ErrBadRequest, "parsing id")
@ -380,6 +389,9 @@ func (s *sqlDatabase) DeleteEntityPool(_ context.Context, entity params.GithubEn
}
func (s *sqlDatabase) UpdateEntityPool(_ context.Context, entity params.GithubEntity, poolID string, param params.UpdatePoolParams) (updatedPool params.Pool, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.PoolEntityType, common.UpdateOperation, updatedPool)

9
database/sql/repositories.go
View file

@ -30,6 +30,9 @@ import (
)
func (s *sqlDatabase) CreateRepository(ctx context.Context, owner, name, credentialsName, webhookSecret string, poolBalancerType params.PoolBalancerType) (param params.Repository, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.RepositoryEntityType, common.CreateOperation, param)
@ -122,6 +125,9 @@ func (s *sqlDatabase) ListRepositories(_ context.Context) ([]params.Repository,
}
func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) (err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
repo, err := s.getRepoByID(ctx, s.conn, repoID, "Endpoint", "Credentials", "Credentials.Endpoint")
if err != nil {
return errors.Wrap(err, "fetching repo")
@ -147,6 +153,9 @@ func (s *sqlDatabase) DeleteRepository(ctx context.Context, repoID string) (err
}
func (s *sqlDatabase) UpdateRepository(ctx context.Context, repoID string, param params.UpdateEntityParams) (newParams params.Repository, err error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
defer func() {
if err == nil {
s.sendNotify(common.RepositoryEntityType, common.UpdateOperation, newParams)

6
database/sql/sql.go
View file

@ -20,6 +20,7 @@ import (
"log/slog"
"net/url"
"strings"
"sync"
"github.com/pkg/errors"
"gorm.io/driver/mysql"
@ -91,6 +92,11 @@ type sqlDatabase struct {
ctx context.Context
cfg config.Database
producer common.Producer
// while busy_timeout helps, in situations of high contention, we can still
// end up with multiple threads trying to write to the database. SQLite does now
// support row level locking.
writeMux sync.Mutex
}
var renameTemplate = `

6
database/sql/users.go
View file

@ -57,6 +57,9 @@ func (s *sqlDatabase) getUserByID(tx *gorm.DB, userID string) (User, error) {
}
func (s *sqlDatabase) CreateUser(_ context.Context, user params.NewUserParams) (params.User, error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
if user.Username == "" || user.Email == "" || user.Password == "" {
return params.User{}, runnerErrors.NewBadRequestError("missing username, password or email")
}
@ -119,6 +122,9 @@ func (s *sqlDatabase) GetUserByID(_ context.Context, userID string) (params.User
}
func (s *sqlDatabase) UpdateUser(_ context.Context, user string, param params.UpdateUserParams) (params.User, error) {
s.writeMux.Lock()
defer s.writeMux.Unlock()
var err error
var dbUser User
err = s.conn.Transaction(func(tx *gorm.DB) error {