From 2ef7c93fee34254f2b9829bd83c5e21db5507ba3 Mon Sep 17 00:00:00 2001 From: "Alessandro (Ale) Segala" <43508+ItalyPaleAle@users.noreply.github.com> Date: Wed, 29 Jul 2026 13:15:40 -0700 Subject: [PATCH] feat: migrate app locks to Francis (#1624) Co-authored-by: Elias Schneider Co-authored-by: Kyle Mendell --- backend/go.mod | 16 +- backend/go.sum | 17 + .../internal/bootstrap/actors_bootstrap.go | 80 +++- backend/internal/bootstrap/bootstrap.go | 36 +- .../bootstrap/e2etest_router_bootstrap.go | 2 +- .../internal/bootstrap/services_bootstrap.go | 2 - backend/internal/cmds/import.go | 115 ++++-- backend/internal/common/env_config.go | 5 + .../internal/controller/e2etest_controller.go | 5 - backend/internal/service/app_lock_service.go | 363 ------------------ .../internal/service/app_lock_service_test.go | 300 --------------- backend/internal/service/e2etest_service.go | 9 +- backend/internal/service/export_service.go | 8 - 13 files changed, 181 insertions(+), 777 deletions(-) delete mode 100644 backend/internal/service/app_lock_service.go delete mode 100644 backend/internal/service/app_lock_service_test.go diff --git a/backend/go.mod b/backend/go.mod index 9768ac89..d427b707 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -24,17 +24,17 @@ require ( github.com/go-webauthn/webauthn v0.17.4 github.com/golang-migrate/migrate/v4 v4.19.1 github.com/google/uuid v1.6.0 - github.com/italypaleale/francis v0.1.0-beta.16 - github.com/italypaleale/go-kit v0.0.0-20260725195228-78f113702f86 + github.com/italypaleale/francis v0.1.0-beta.17 + github.com/italypaleale/go-kit v0.0.0-20260729024436-06ef4dfe084c github.com/italypaleale/go-sql-utils v0.2.4 github.com/jackc/pgx/v5 v5.10.0 github.com/jinzhu/copier v0.4.0 github.com/joho/godotenv v1.5.1 github.com/lestrrat-go/httprc/v3 v3.0.6 github.com/lestrrat-go/jwx/v3 v3.1.1 - github.com/libtnb/sqlite v1.2.0 + github.com/libtnb/sqlite v1.2.1 github.com/lmittmann/tint v1.2.0 - github.com/mattn/go-isatty v0.0.23 + github.com/mattn/go-isatty v0.0.24 github.com/mileusna/useragent v1.3.5 github.com/orandin/slog-gorm v1.4.0 github.com/ory/fosite v0.49.1-0.20250703093431-a5f0b09bf31c @@ -147,7 +147,7 @@ require ( github.com/jinzhu/now v1.1.5 // indirect github.com/jonboulle/clockwork v0.5.0 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.19.1 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/lestrrat-go/blackmagic v1.0.4 // indirect @@ -181,8 +181,8 @@ require ( github.com/prometheus/otlptranslator v1.0.0 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/quic-go/qpack v0.6.0 // indirect - github.com/quic-go/quic-go v0.60.0 // indirect - github.com/quic-go/webtransport-go v0.11.1 // indirect + github.com/quic-go/quic-go v0.61.0 // indirect + github.com/quic-go/webtransport-go v0.12.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/robfig/cron/v3 v3.0.1 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect @@ -243,7 +243,7 @@ require ( gorm.io/driver/clickhouse v0.7.0 // indirect gorm.io/driver/mysql v1.5.7 // indirect k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3 // indirect - modernc.org/libc v1.74.1 // indirect + modernc.org/libc v1.74.4 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect ) diff --git a/backend/go.sum b/backend/go.sum index 5ab2f3ae..896eb54c 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -265,8 +265,12 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/italypaleale/francis v0.1.0-beta.16 h1:bkt+8iA2f/hi14vve0cYaUMvVhHBIoD/cLVlUkHR69c= github.com/italypaleale/francis v0.1.0-beta.16/go.mod h1:KKwS+57OBD/MoHBVfbbMelA2vUx65fPiG9fhdWWapFc= +github.com/italypaleale/francis v0.1.0-beta.17 h1:woN/6DE98iXkdjEqxPdcMJ12EqFR+U5MFlB51qza1ZY= +github.com/italypaleale/francis v0.1.0-beta.17/go.mod h1:C3CXbZt+wVxbSG8truMlaaEaB4W/wbx2sV89owwoGJE= github.com/italypaleale/go-kit v0.0.0-20260725195228-78f113702f86 h1:719T7W8hLVjelch856Sern60QAPMn0fIE87i91YVcfw= github.com/italypaleale/go-kit v0.0.0-20260725195228-78f113702f86/go.mod h1:0Sy3bN3qnSy2kgcJ05A2CsP6os5wmLC9lPraFr+0jGk= +github.com/italypaleale/go-kit v0.0.0-20260729024436-06ef4dfe084c h1:7mIVxDhwU5nwZI20tTXhLqjJvGIWdY8UaBUCiPZrfmo= +github.com/italypaleale/go-kit v0.0.0-20260729024436-06ef4dfe084c/go.mod h1:CSkdRl2TNG0yCSwsuqMD4pI+biKiFNIxfe6c8QdocjM= github.com/italypaleale/go-sql-utils v0.2.4 h1:6CN8y3qEdNzvYlS/JK6N65E8cL9F8a6OBCJjzaQIv3c= github.com/italypaleale/go-sql-utils v0.2.4/go.mod h1:BJStxMfB6fzYVcOe0oZQCjGIPZQu76UBmg1Wuy6Z/7I= github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 h1:D/V0gu4zQ3cL2WKeVNVM4r2gLxGGf6McLwgXzRTo2RQ= @@ -311,6 +315,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo= @@ -350,12 +356,16 @@ github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= github.com/libtnb/sqlite v1.2.0 h1:XsA7jsXHH2qmFkTWoy5YCKJybzHzWQ0flqDFP5Y9Yto= github.com/libtnb/sqlite v1.2.0/go.mod h1:O6vURH5fa5IgSmXd/qLAL2zSnYFUd7xSIumsmX3BrSI= +github.com/libtnb/sqlite v1.2.1 h1:lrhN3yG49pj1gmtMMP6Es0lhKIkzDeqBul2EWOLyjpY= +github.com/libtnb/sqlite v1.2.1/go.mod h1:pGq13BirCjkM6ihkWWxs0PGFhBiwUyA0PqdSmgDWfmM= github.com/lmittmann/tint v1.2.0 h1:AogHRHy8HUJUnNJBHJlYa+fR4YY8mko2cnCp67xn9JY= github.com/lmittmann/tint v1.2.0/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-isatty v0.0.23 h1:cYwCQTQf3HB6xUC+BtyCLZNr7IzbOmoZbmssVNzSyiQ= github.com/mattn/go-isatty v0.0.23/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= +github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= github.com/mattn/go-sqlite3 v1.14.42 h1:MigqEP4ZmHw3aIdIT7T+9TLa90Z6smwcthx+Azv4Cgo= github.com/mattn/go-sqlite3 v1.14.42/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= github.com/mattn/goveralls v0.0.12 h1:PEEeF0k1SsTjOBQ8FOmrOAoCu4ytuMaWCnWe94zxbCg= @@ -446,8 +456,12 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.60.0 h1:xcQioE8OM66UQLeUMHltK1CCcOu3JbVB4JAQdDQSB+0= github.com/quic-go/quic-go v0.60.0/go.mod h1:wpKpjmPpftl30sL6pFh7REVpjbcCVy4zt2vDyK1TuJk= +github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA= +github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs= github.com/quic-go/webtransport-go v0.11.1 h1:rrFQMO+7/52ZDJ04fsrjIaWqn6q1z1MYo9iVFq6JtbA= github.com/quic-go/webtransport-go v0.11.1/go.mod h1:SHgEzUFVyj+9WUSuGB1P6Zd351Pww2leWV3SwlTovkA= +github.com/quic-go/webtransport-go v0.12.0 h1:CpnKNwZvdV0LD73xoHO8QaR0NI3llqpWRwnazdZS0sE= +github.com/quic-go/webtransport-go v0.12.0/go.mod h1:GHne8aRFJ24h73pAMrcywXtuaz/ShBXCLXLvG/NPFdU= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= @@ -771,6 +785,7 @@ k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3 h1:jVkFFVfXdXP74B/zbO3hM3hpSFD0x k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3/go.mod h1:M2s5JB1lIYP3jzZdorPLHXIPJzt9vv2muW5a6L9DtNM= modernc.org/cc/v4 v4.29.0 h1:CXgwL8cvxmyzBQZzbSl/6xFtMCryb6u8IOqDci39cgc= modernc.org/cc/v4 v4.29.0/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/cc/v4 v4.29.1 h1:MKgdCV3WykTSPqpVrnxdEDS0HEd2FHpKZDzxzU5LyeI= modernc.org/ccgo/v4 v4.34.6 h1:sBgfIwyN0TQ9C5hwIeuqyeAKyMWnbvj2fvpF4L11uzU= modernc.org/ccgo/v4 v4.34.6/go.mod h1:SZ8YcN9NG7XVsQYdm6jYBvi8PQP1qi+kqB6OhjqI3Fk= modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= @@ -783,6 +798,8 @@ modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= modernc.org/libc v1.74.1 h1:bdR4VTKFMC4966QSNZ05XLGI/VwzVa2kTUX51Dm0riQ= modernc.org/libc v1.74.1/go.mod h1:uH4t5bOx3G3g9Xcmj10YKlTcVISlRDwv8VoQJG9n8Os= +modernc.org/libc v1.74.4 h1:fX1Omw4o2/1C2iRkkIsrQTasJQldLhRmuPreXLoWs9k= +modernc.org/libc v1.74.4/go.mod h1:eeQAS9W3sZeKYMFubydxJpII9ybHWshk+7or7bLG9co= modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= diff --git a/backend/internal/bootstrap/actors_bootstrap.go b/backend/internal/bootstrap/actors_bootstrap.go index abdd833e..01114fbe 100644 --- a/backend/internal/bootstrap/actors_bootstrap.go +++ b/backend/internal/bootstrap/actors_bootstrap.go @@ -10,6 +10,7 @@ import ( "time" "github.com/italypaleale/francis/builtin/ratelimit" + "github.com/italypaleale/francis/components" "github.com/italypaleale/francis/components/postgres" "github.com/italypaleale/francis/host/local" "github.com/jackc/pgx/v5/pgxpool" @@ -43,22 +44,35 @@ func NewActors(o NewActorsOpts) (*local.Host, map[string]*ratelimit.RateLimitSer return nil, nil, fmt.Errorf("failed to derive PSK: %w", err) } + // Derive the cluster host limit from the HA setting + // With HA disabled the cluster is capped at a single replica + maxHosts := 1 + if o.EnvConfig.HAEnabled { + // 0 = no cap + maxHosts = 0 + } + // Options for the host opts := []local.HostOption{ local.WithAddress(net.JoinHostPort(o.EnvConfig.ActorsHost, o.EnvConfig.ActorsPort)), local.WithLogger(log.With("scope", "actor-host")), local.WithRuntimePSKs(psk), local.WithShutdownGracePeriod(10 * time.Second), - // TODO: Tweak these values once Pocket ID fully supports horizontal scaling. - // The relaxed intervals are appropriate for a single active host, but should be - // tuned for lower latency and better distribution across a multi-host cluster. - local.WithHostHealthCheckDeadline(90 * time.Second), - local.WithAlarmsPollInterval(5 * time.Minute), - local.WithAlarmsFetchAheadInterval(5 * time.Minute), + local.WithMaxHosts(maxHosts), + local.WithHostHealthCheckDeadline(ActorsHostHealthCheckDeadline(o.EnvConfig.HAEnabled)), + } + + // With a single active host the relaxed alarm intervals reduce database load + // When HA is enabled they are dropped so Francis uses its tighter defaults, which distribute alarm work and fail over faster across multiple hosts + if !o.EnvConfig.HAEnabled { + opts = append(opts, + local.WithAlarmsPollInterval(5*time.Minute), + local.WithAlarmsFetchAheadInterval(5*time.Minute), + ) } // Add the database connection - providerOpt, err := o.getProvider() + providerOpt, err := o.getProviderOption() if err != nil { return nil, nil, err } @@ -111,7 +125,7 @@ func NewActorStateStore(db *gorm.DB, pg *pgxpool.Pool) (*local.Host, error) { opts.SQLite = sqlDB } - providerOpt, err := opts.getProvider() + providerOpt, err := opts.getProviderOption() if err != nil { return nil, err } @@ -126,23 +140,53 @@ func NewActorStateStore(db *gorm.DB, pg *pgxpool.Pool) (*local.Host, error) { ) } -func (o *NewActorsOpts) getProvider() (local.HostOption, error) { +// ActorsHostHealthCheckDeadline returns the health-check deadline the actor host uses for the given HA setting +// This is exported because the import method needs it too +func ActorsHostHealthCheckDeadline(haEnabled bool) time.Duration { + if haEnabled { + return components.DefaultHostHealthCheckDeadline + } + + // A single active host does not need aggressive health checks, so a longer deadline reduces database load + return 90 * time.Second +} + +// ActorsProviderOptions builds the Francis provider options for the given database handles +// The actor host and the cluster admin must use the same options so they address the same cluster +// This is implemented separately and exported because the import method needs it too +func ActorsProviderOptions(pg *pgxpool.Pool, sqliteDB *sql.DB) (components.ProviderOptions, error) { switch { - case o.Postgres != nil && o.SQLite != nil: + case pg != nil && sqliteDB != nil: return nil, errors.New("cannot have both Postgres and SQLite connections") - case o.Postgres != nil: - return local.WithPostgresProvider(postgres.PostgresProviderOptions{ - DB: o.Postgres, - }), nil - case o.SQLite != nil: - return local.WithSQLiteProvider(local.SQLiteProviderOptions{ - DB: o.SQLite, - }), nil + case pg != nil: + return postgres.PostgresProviderOptions{ + DB: pg, + }, nil + case sqliteDB != nil: + return local.SQLiteProviderOptions{ + DB: sqliteDB, + }, nil default: return nil, errors.New("one of Postgres and SQLite must be set") } } +// getProviderOption wraps the shared provider options in the host option the local host expects +func (o *NewActorsOpts) getProviderOption() (local.HostOption, error) { + providerOpts, err := ActorsProviderOptions(o.Postgres, o.SQLite) + if err != nil { + return nil, err + } + switch v := providerOpts.(type) { + case postgres.PostgresProviderOptions: + return local.WithPostgresProvider(v), nil + case local.SQLiteProviderOptions: + return local.WithSQLiteProvider(v), nil + default: + return nil, fmt.Errorf("unsupported provider options type: %T", providerOpts) + } +} + func (o *NewActorsOpts) registerCronJobs(host *local.Host) (err error) { // In test mode, we do not register anything if common.EnvConfig.AppEnv == "test" { diff --git a/backend/internal/bootstrap/bootstrap.go b/backend/internal/bootstrap/bootstrap.go index a69c8ac9..6581dc04 100644 --- a/backend/internal/bootstrap/bootstrap.go +++ b/backend/internal/bootstrap/bootstrap.go @@ -9,6 +9,7 @@ import ( _ "github.com/golang-migrate/migrate/v4/source/file" + "github.com/italypaleale/francis/components" "github.com/italypaleale/francis/host/local" "github.com/italypaleale/go-kit/servicerunner" "gorm.io/gorm" @@ -16,7 +17,6 @@ import ( "github.com/pocket-id/pocket-id/backend/internal/common" "github.com/pocket-id/pocket-id/backend/internal/instanceid" "github.com/pocket-id/pocket-id/backend/internal/job" - "github.com/pocket-id/pocket-id/backend/internal/service" "github.com/pocket-id/pocket-id/backend/internal/storage" ) @@ -43,7 +43,7 @@ func Bootstrap(ctx context.Context) error { } if pg != nil { defer func() { - // Close the database connection pool only after the shutdown functions have run: some of them (e.g. releasing the application lock) still need to query the database. + // Close the database connection pool only after the shutdown functions have run: some of them (e.g. the actor host deregistering itself from the cluster) still need to query the database. pg.Close() }() } @@ -112,40 +112,19 @@ func Bootstrap(ctx context.Context) error { if err != nil { return fmt.Errorf("failed to initialize services: %w", err) } - services = append(services, svc.appLockService.RunRenewal) // Migrate the pre-actor signup tokens into their actors, once the actor host is ready services = append(services, actorsReady.Await(svc.userSignUpModule.RunSignupTokenMigration)) - // Acquire the lock from the app lock service - waitUntil, err := svc.appLockService.Acquire(ctx, false) - if errors.Is(err, service.ErrLockUnavailable) { - return errors.New("it appears that there's already one instance of Pocket ID running; running multiple replicas of Pocket ID is currently not supported") - } else if err != nil { - return fmt.Errorf("failed to acquire application lock: %w", err) - } - - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(time.Until(waitUntil)): - } - - shutdowns.Add(func(shutdownCtx context.Context) error { - sErr := svc.appLockService.Release(shutdownCtx) - if sErr != nil { - return fmt.Errorf("failed to release application lock: %w", sErr) - } - return nil - }) - // Register scheduled jobs, only in non-test mode if common.EnvConfig.AppEnv != "test" { err = registerScheduledJobs(ctx, db, svc, scheduler) if err != nil { return fmt.Errorf("failed to register scheduled jobs: %w", err) } - services = append(services, scheduler.Run) + + // The scheduler must wait on the actor host being ready, since jobs invoke actors + services = append(services, actorsReady.Await(scheduler.Run)) } // Init the router @@ -162,7 +141,10 @@ func Bootstrap(ctx context.Context) error { // Run all background services // This call blocks until the context is canceled err = servicerunner.NewServiceRunner(services...).Run(ctx) - if err != nil { + if errors.Is(err, components.ErrClusterFull) { + // TODO: Once HA mode is supported, add a note about enabling it + return errors.New("it appears that there's already one instance of Pocket ID running - running multiple replicas is not (yet) supported") + } else if err != nil { return fmt.Errorf("failed to run services: %w", err) } diff --git a/backend/internal/bootstrap/e2etest_router_bootstrap.go b/backend/internal/bootstrap/e2etest_router_bootstrap.go index db1eb663..d5c99d1c 100644 --- a/backend/internal/bootstrap/e2etest_router_bootstrap.go +++ b/backend/internal/bootstrap/e2etest_router_bootstrap.go @@ -17,7 +17,7 @@ import ( func init() { registerTestControllers = []func(apiGroup *gin.RouterGroup, db *gorm.DB, svc *services){ func(apiGroup *gin.RouterGroup, db *gorm.DB, svc *services) { - testService, err := service.NewTestService(db, svc.actors, svc.appConfigService, svc.jwtService, svc.ldapService, svc.appLockService, svc.fileStorage) + testService, err := service.NewTestService(db, svc.actors, svc.appConfigService, svc.jwtService, svc.ldapService, svc.fileStorage) if err != nil { slog.Error("Failed to initialize test service", slog.Any("error", err)) os.Exit(1) diff --git a/backend/internal/bootstrap/services_bootstrap.go b/backend/internal/bootstrap/services_bootstrap.go index 1974de78..3b02c03c 100644 --- a/backend/internal/bootstrap/services_bootstrap.go +++ b/backend/internal/bootstrap/services_bootstrap.go @@ -38,7 +38,6 @@ type services struct { ldapService *service.LdapService versionService *service.VersionService fileStorage storage.FileStorage - appLockService *service.AppLockService apiKeyModule *apikey.Module deviceLoginModule *devicelogin.Module @@ -74,7 +73,6 @@ func initServices( svc.fileStorage = fileStorage svc.appImagesService = service.NewAppImagesService(imageExtensions, fileStorage) - svc.appLockService = service.NewAppLockService(db) svc.emailModule, err = email.New(db) if err != nil { diff --git a/backend/internal/cmds/import.go b/backend/internal/cmds/import.go index 40273701..5cd5f1ff 100644 --- a/backend/internal/cmds/import.go +++ b/backend/internal/cmds/import.go @@ -3,15 +3,18 @@ package cmds import ( "archive/zip" "context" + "database/sql" "errors" "fmt" "io" + "log/slog" "os" "path/filepath" "time" + "github.com/italypaleale/francis/clusteradmin" + "github.com/italypaleale/francis/components" "github.com/spf13/cobra" - "gorm.io/gorm" "github.com/pocket-id/pocket-id/backend/internal/bootstrap" "github.com/pocket-id/pocket-id/backend/internal/common" @@ -38,7 +41,7 @@ func init() { importCmd.Flags().StringVarP(&flags.Path, "path", "p", "pocket-id-export.zip", "Path to the ZIP file to import the data from, or '-' to read from stdin") importCmd.Flags().BoolVarP(&flags.Yes, "yes", "y", false, "Skip confirmation prompts") - importCmd.Flags().BoolVarP(&flags.ForcefullyAcquireLock, "forcefully-acquire-lock", "", false, "Forcefully acquire the application lock by terminating the Pocket ID instance") + importCmd.Flags().BoolVarP(&flags.ForcefullyAcquireLock, "forcefully-acquire-lock", "", false, "Forcefully acquire exclusive access by terminating any running Pocket ID instance") rootCmd.AddCommand(importCmd) } @@ -73,17 +76,45 @@ func runImport(ctx context.Context, flags importFlags) error { } defer zipReader.Close() - db, _, err := bootstrap.ConnectDatabase(ctx) + // Connect to the database without running migrations: the import re-creates the Pocket ID schema itself + db, pg, err := bootstrap.ConnectDatabase(ctx) if err != nil { return err } - err = acquireImportLock(ctx, db, flags.ForcefullyAcquireLock) + // The cluster admin talks to the same database as the actor host, so build its provider options the same way the host does + var sqliteDB *sql.DB + if pg == nil { + sqliteDB, err = db.DB() + if err != nil { + return fmt.Errorf("failed to get sql.DB connection: %w", err) + } + } + providerOpts, err := bootstrap.ActorsProviderOptions(pg, sqliteDB) if err != nil { return err } - storage, err := bootstrap.InitStorage(ctx, db) + // Take exclusive access to the cluster so no Pocket ID replica is running while we overwrite the database + release, lost, err := acquireExclusiveAccess(ctx, providerOpts, flags.ForcefullyAcquireLock) + if err != nil { + return err + } + defer release() + + // Abort the import if exclusive access is lost partway through (for example if the lease can no longer be renewed) + importCtx, cancel := context.WithCancel(ctx) + defer cancel() + go func() { + select { + case <-lost: + cancel() + case <-importCtx.Done(): + } + }() + + // Init the storage provider + storage, err := bootstrap.InitStorage(importCtx, db) if err != nil { return fmt.Errorf("failed to initialize storage: %w", err) } @@ -93,8 +124,11 @@ func runImport(ctx context.Context, flags importFlags) error { _ = storage.Close() }() + // Create the import service importService := service.NewImportService(db, storage) - err = importService.ImportFromZip(ctx, &zipReader.Reader) + + // Load from ZIP + err = importService.ImportFromZip(importCtx, &zipReader.Reader) if err != nil { return fmt.Errorf("failed to import data from zip: %w", err) } @@ -103,41 +137,46 @@ func runImport(ctx context.Context, flags importFlags) error { return nil } -func acquireImportLock(ctx context.Context, db *gorm.DB, force bool) error { - // Check if the kv table exists, in case we are starting from an empty database - exists, err := utils.DBTableExists(db, "kv") +// acquireExclusiveAccess takes an exclusive-access lease on the cluster so the import can safely overwrite the database. +// +// It returns a release function that must be called once the import is done, and a channel that is closed if the lease is lost while it is held. +func acquireExclusiveAccess(ctx context.Context, providerOpts components.ProviderOptions, force bool) (release func(), lost <-chan struct{}, err error) { + // New initializes the provider, applying the actor host's schema migrations, so this also works against a brand-new (empty) database + admin, err := clusteradmin.New(ctx, providerOpts, clusteradmin.Options{ + // Match the actor host so the admin waits the right amount of time for hosts to drain + HostHealthCheckDeadline: bootstrap.ActorsHostHealthCheckDeadline(common.EnvConfig.HAEnabled), + Logger: slog.Default(), + }) if err != nil { - return fmt.Errorf("failed to check if kv table exists: %w", err) - } - if !exists { - // This either means the database is empty, or the import is into an old version of PocketID that doesn't support locks - // In either case, there's no lock to acquire - fmt.Println("Could not acquire a lock because the 'kv' table does not exist. This is fine if you're importing into a new database, but make sure that there isn't an instance of Pocket ID currently running and using the same database.") - return nil + return nil, nil, fmt.Errorf("failed to create cluster admin: %w", err) } - // Note that we do not call a deferred Release if the data was imported - // This is because we are overriding the contents of the database, so the lock is automatically lost - appLockService := service.NewAppLockService(db) - - opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) - defer cancel() - - waitUntil, err := appLockService.Acquire(opCtx, force) - if errors.Is(err, service.ErrLockUnavailable) { - //nolint:staticcheck - return errors.New("Pocket ID must be stopped before importing data; please stop the running instance or run with --forcefully-acquire-lock to terminate the other instance") - } else if err != nil { - return fmt.Errorf("failed to acquire application lock: %w", err) + lost, err = admin.AcquireExclusive(ctx, clusteradmin.AcquireOptions{Force: force}) + if err != nil { + _ = admin.Close() + switch { + case errors.Is(err, components.ErrHostsConnected): + //nolint:staticcheck + return nil, nil, errors.New("Pocket ID must be stopped before importing data - please stop the running instance or run with --forcefully-acquire-lock to terminate the other instance") + case errors.Is(err, components.ErrExclusiveHeld): + return nil, nil, errors.New("another exclusive operation, such as another import, is already in progress; please wait for it to complete and try again") + default: + return nil, nil, fmt.Errorf("failed to acquire exclusive access: %w", err) + } } - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(time.Until(waitUntil)): + release = func() { + // The import preserves the actor host's "francis_" tables, including the lease row, so the lease must be released explicitly + // Detach from ctx so the release still runs even if the import was canceled + releaseCtx, cancelRelease := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cancelRelease() + rErr := admin.ReleaseExclusive(releaseCtx) + if rErr != nil { + slog.WarnContext(ctx, "Failed to release exclusive access", slog.Any("error", rErr)) + } + _ = admin.Close() } - - return nil + return release, lost, nil } func askForConfirmation() (bool, error) { @@ -174,13 +213,15 @@ func readZipFromStdin() (*zip.ReadCloser, func(), error) { _ = os.Remove(tmpFile.Name()) } - if _, err := io.Copy(tmpFile, os.Stdin); err != nil { + _, err = io.Copy(tmpFile, os.Stdin) + if err != nil { tmpFile.Close() cleanup() return nil, nil, fmt.Errorf("failed to read data from stdin: %w", err) } - if err := tmpFile.Close(); err != nil { + err = tmpFile.Close() + if err != nil { cleanup() return nil, nil, fmt.Errorf("failed to close temporary file: %w", err) } diff --git a/backend/internal/common/env_config.go b/backend/internal/common/env_config.go index 578b6155..f793eed2 100644 --- a/backend/internal/common/env_config.go +++ b/backend/internal/common/env_config.go @@ -85,6 +85,11 @@ type EnvConfigSchema struct { ActorsPort string `env:"ACTORS_PORT"` ActorsHost string `env:"ACTORS_HOST" options:"toLower"` + // HAEnabled turns on high-availability mode, allowing more than one replica of Pocket ID to run against the same database at once + // It is intentionally not bound to an environment variable while HA support is still being completed + // TODO: Add env var when HA mode is ready + HAEnabled bool + LogLevel string `env:"LOG_LEVEL" options:"toLower"` LogJSON bool `env:"LOG_JSON"` } diff --git a/backend/internal/controller/e2etest_controller.go b/backend/internal/controller/e2etest_controller.go index d15bfbfb..2c6abc73 100644 --- a/backend/internal/controller/e2etest_controller.go +++ b/backend/internal/controller/e2etest_controller.go @@ -41,11 +41,6 @@ func (tc *TestController) resetAndSeedHandler(c *gin.Context) { return } - if err := tc.TestService.ResetLock(c.Request.Context()); err != nil { - _ = c.Error(err) - return - } - if err := tc.TestService.ResetApplicationImages(c.Request.Context()); err != nil { _ = c.Error(err) return diff --git a/backend/internal/service/app_lock_service.go b/backend/internal/service/app_lock_service.go deleted file mode 100644 index c8f62d92..00000000 --- a/backend/internal/service/app_lock_service.go +++ /dev/null @@ -1,363 +0,0 @@ -package service - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "log/slog" - "os" - "time" - - "github.com/google/uuid" - "go.opentelemetry.io/otel/trace" - "gorm.io/gorm" - "gorm.io/gorm/clause" - - "github.com/pocket-id/pocket-id/backend/internal/model" - "github.com/pocket-id/pocket-id/backend/internal/tracing" -) - -var ( - ErrLockUnavailable = errors.New("lock is already held by another process") - ErrLockLost = errors.New("lock ownership lost") -) - -const ( - ttl = 30 * time.Second - renewInterval = 20 * time.Second - renewRetries = 3 - lockKey = "application_lock" -) - -type AppLockService struct { - db *gorm.DB - lockID string - processID int64 - hostID string -} - -func NewAppLockService(db *gorm.DB) *AppLockService { - host, err := os.Hostname() - if err != nil || host == "" { - host = "unknown-host" - } - - return &AppLockService{ - db: db, - processID: int64(os.Getpid()), - hostID: host, - lockID: uuid.NewString(), - } -} - -type lockValue struct { - ProcessID int64 `json:"process_id"` - HostID string `json:"host_id"` - LockID string `json:"lock_id"` - ExpiresAt int64 `json:"expires_at"` -} - -func (lv *lockValue) Marshal() (string, error) { - data, err := json.Marshal(lv) - if err != nil { - return "", err - } - return string(data), nil -} - -func (lv *lockValue) Unmarshal(raw string) error { - if raw == "" { - return nil - } - return json.Unmarshal([]byte(raw), lv) -} - -// Acquire obtains the lock. When force is true, the lock is stolen from any existing owner. -// If the lock is forcefully acquired, it blocks until the previous lock has expired. -func (s *AppLockService) Acquire(ctx context.Context, force bool) (waitUntil time.Time, err error) { - ctx, span := tracing.Start(ctx, "pocketid.applock.acquire", trace.WithSpanKind(trace.SpanKindInternal)) - waitUntil, err = s.acquire(ctx, force) - if err != nil { - tracing.End(span, err) - return waitUntil, err - } - - span.End() - return waitUntil, nil -} - -func (s *AppLockService) acquire(ctx context.Context, force bool) (waitUntil time.Time, err error) { - tx := s.db.WithContext(ctx).Begin() - err = tx.Error - if err != nil { - return time.Time{}, fmt.Errorf("begin lock transaction: %w", err) - } - defer tx.Rollback() - - var prevLockRaw string - err = tx. - WithContext(ctx). - Model(&model.KV{}). - Where("key = ?", lockKey). - Clauses(clause.Locking{Strength: "UPDATE"}). - Select("value"). - Scan(&prevLockRaw). - Error - if err != nil { - return time.Time{}, fmt.Errorf("query existing lock: %w", err) - } - - var prevLock lockValue - if prevLockRaw != "" { - err = prevLock.Unmarshal(prevLockRaw) - if err != nil { - return time.Time{}, fmt.Errorf("decode existing lock value: %w", err) - } - } - - now := time.Now() - nowUnix := now.Unix() - - value := lockValue{ - ProcessID: s.processID, - HostID: s.hostID, - LockID: s.lockID, - ExpiresAt: now.Add(ttl).Unix(), - } - raw, err := value.Marshal() - if err != nil { - return time.Time{}, fmt.Errorf("encode lock value: %w", err) - } - - var query string - switch s.db.Name() { - case "sqlite": - query = ` - INSERT INTO kv (key, value) - VALUES (?, ?) - ON CONFLICT(key) DO UPDATE SET - value = excluded.value - WHERE (json_extract(kv.value, '$.expires_at') < ?) OR ? - ` - case "postgres": - query = ` - INSERT INTO kv (key, value) - VALUES ($1, $2) - ON CONFLICT(key) DO UPDATE SET - value = excluded.value - WHERE ((kv.value::json->>'expires_at')::bigint < $3) OR ($4::boolean IS TRUE) - ` - default: - return time.Time{}, fmt.Errorf("unsupported database dialect: %s", s.db.Name()) - } - - res := tx.WithContext(ctx).Exec(query, lockKey, raw, nowUnix, force) - if res.Error != nil { - return time.Time{}, fmt.Errorf("lock acquisition failed: %w", res.Error) - } - - err = tx.Commit().Error - if err != nil { - return time.Time{}, fmt.Errorf("commit lock acquisition: %w", err) - } - - // If there is a lock that is not expired and force is false, no rows will be affected - if res.RowsAffected == 0 { - err = ErrLockUnavailable - return time.Time{}, err - } - - if force && prevLock.ExpiresAt > nowUnix && prevLock.LockID != s.lockID { - waitUntil = time.Unix(prevLock.ExpiresAt, 0) - } - - attrs := []any{ - slog.Int64("process_id", s.processID), - slog.String("host_id", s.hostID), - } - wait := time.Until(waitUntil) - if wait > 0 { - attrs = append(attrs, slog.Duration("wait_before_proceeding", wait)) - } - slog.InfoContext(ctx, "Acquired application lock", attrs...) - - return waitUntil, nil -} - -// RunRenewal keeps renewing the lock until the context is canceled. -func (s *AppLockService) RunRenewal(ctx context.Context) error { - ticker := time.NewTicker(renewInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return nil - case <-ticker.C: - renewCtx, span := tracing.Start(ctx, "pocketid.applock.renew", trace.WithSpanKind(trace.SpanKindInternal)) - err := s.renew(renewCtx) - tracing.End(span, err) - if err != nil { - return fmt.Errorf("renew lock: %w", err) - } - } - } -} - -// Release releases the lock if it is held by this process. -func (s *AppLockService) Release(ctx context.Context) (err error) { - ctx, span := tracing.Start(ctx, "pocketid.applock.release", trace.WithSpanKind(trace.SpanKindInternal)) - err = s.release(ctx) - if err != nil { - tracing.End(span, err) - return err - } - - span.End() - return nil -} - -func (s *AppLockService) release(ctx context.Context) error { - db, err := s.db.DB() - if err != nil { - return fmt.Errorf("failed to get DB connection: %w", err) - } - - var query string - switch s.db.Name() { - case "sqlite": - query = ` -DELETE FROM kv -WHERE key = ? - AND json_extract(value, '$.lock_id') = ? -` - case "postgres": - query = ` -DELETE FROM kv -WHERE key = $1 - AND value::json->>'lock_id' = $2 -` - default: - return fmt.Errorf("unsupported database dialect: %s", s.db.Name()) - } - - opCtx, cancel := context.WithTimeout(ctx, 3*time.Second) - defer cancel() - - res, err := db.ExecContext(opCtx, query, lockKey, s.lockID) - if err != nil { - return fmt.Errorf("release lock failed: %w", err) - } - - count, err := res.RowsAffected() - if err != nil { - return fmt.Errorf("failed to count affected rows: %w", err) - } - - if count == 0 { - // Treat this as a warning, not an error - slog.WarnContext(ctx, "Application lock not held by this process, cannot release", - slog.Int64("process_id", s.processID), - slog.String("host_id", s.hostID), - ) - return nil - } - - slog.InfoContext(ctx, "Released application lock", - slog.Int64("process_id", s.processID), - slog.String("host_id", s.hostID), - ) - return nil -} - -// renew tries to renew the lock, retrying up to renewRetries times (sleeping 1s between attempts). -func (s *AppLockService) renew(ctx context.Context) error { - db, err := s.db.DB() - if err != nil { - return fmt.Errorf("failed to get DB connection: %w", err) - } - - var lastErr error - for attempt := 1; attempt <= renewRetries; attempt++ { - now := time.Now() - nowUnix := now.Unix() - expiresAt := now.Add(ttl).Unix() - - value := lockValue{ - LockID: s.lockID, - ProcessID: s.processID, - HostID: s.hostID, - ExpiresAt: expiresAt, - } - raw, err := value.Marshal() - if err != nil { - return fmt.Errorf("encode lock value: %w", err) - } - - var query string - switch s.db.Name() { - case "sqlite": - query = ` -UPDATE kv -SET value = ? -WHERE key = ? - AND json_extract(value, '$.lock_id') = ? - AND json_extract(value, '$.expires_at') > ? -` - case "postgres": - query = ` -UPDATE kv -SET value = $1 -WHERE key = $2 - AND value::json->>'lock_id' = $3 - AND ((value::json->>'expires_at')::bigint > $4) -` - default: - return fmt.Errorf("unsupported database dialect: %s", s.db.Name()) - } - - opCtx, cancel := context.WithTimeout(ctx, 3*time.Second) - res, err := db.ExecContext(opCtx, query, raw, lockKey, s.lockID, nowUnix) - cancel() - - // Query succeeded, but may have updated 0 rows - if err == nil { - count, err := res.RowsAffected() - if err != nil { - return fmt.Errorf("failed to count affected rows: %w", err) - } - - // If no rows were updated, we lost the lock - if count == 0 { - return ErrLockLost - } - - // All good - slog.DebugContext(ctx, "Renewed application lock", - slog.Int64("process_id", s.processID), - slog.String("host_id", s.hostID), - slog.Duration("duration", time.Since(now)), - ) - return nil - } - - // If we're here, we have an error that can be retried - slog.DebugContext(ctx, "Application lock renewal attempt failed", - slog.Any("error", err), - slog.Duration("duration", time.Since(now)), - ) - lastErr = fmt.Errorf("lock renewal failed: %w", err) - - // Wait before next attempt or cancel if context is done - if attempt < renewRetries { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(1 * time.Second): - } - } - } - - return lastErr -} diff --git a/backend/internal/service/app_lock_service_test.go b/backend/internal/service/app_lock_service_test.go deleted file mode 100644 index 917fe6fb..00000000 --- a/backend/internal/service/app_lock_service_test.go +++ /dev/null @@ -1,300 +0,0 @@ -package service - -import ( - "context" - "testing" - "time" - - "github.com/stretchr/testify/require" - "gorm.io/gorm" - - "github.com/pocket-id/pocket-id/backend/internal/model" - testutils "github.com/pocket-id/pocket-id/backend/internal/utils/testing" -) - -func newTestAppLockService(t *testing.T, db *gorm.DB) *AppLockService { - t.Helper() - - return &AppLockService{ - db: db, - processID: 1, - hostID: "test-host", - lockID: "a13c7673-c7ae-49f1-9112-2cd2d0d4b0c1", - } -} - -func insertLock(t *testing.T, db *gorm.DB, value lockValue) { - t.Helper() - - raw, err := value.Marshal() - require.NoError(t, err) - - err = db.Create(&model.KV{Key: lockKey, Value: &raw}).Error - require.NoError(t, err) -} - -func readLockValue(t *testing.T, db *gorm.DB) lockValue { - t.Helper() - - var row model.KV - err := db.Take(&row, "key = ?", lockKey).Error - require.NoError(t, err) - - require.NotNil(t, row.Value) - - var value lockValue - err = value.Unmarshal(*row.Value) - require.NoError(t, err) - - return value -} - -func lockDatabaseForWrite(t *testing.T, db *gorm.DB) *gorm.DB { - t.Helper() - - tx := db.Begin() - require.NoError(t, tx.Error) - - // Keep a write transaction open to block other queries. - err := tx.Exec( - `INSERT INTO kv (key, value) VALUES (?, ?) ON CONFLICT(key) DO NOTHING`, - lockKey, - `{"expires_at":0}`, - ).Error - require.NoError(t, err) - - return tx -} - -func TestAppLockServiceAcquire(t *testing.T) { - t.Run("creates new lock when none exists", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - stored := readLockValue(t, db) - require.Equal(t, service.processID, stored.ProcessID) - require.Equal(t, service.hostID, stored.HostID) - require.Greater(t, stored.ExpiresAt, time.Now().Unix()) - }) - - t.Run("returns ErrLockUnavailable when lock held by another process", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - existing := lockValue{ - ProcessID: 99, - HostID: "other-host", - ExpiresAt: time.Now().Add(ttl).Unix(), - } - insertLock(t, db, existing) - - _, err := service.Acquire(context.Background(), false) - require.ErrorIs(t, err, ErrLockUnavailable) - - current := readLockValue(t, db) - require.Equal(t, existing, current) - }) - - t.Run("force acquisition steals lock", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - insertLock(t, db, lockValue{ - ProcessID: 99, - HostID: "other-host", - ExpiresAt: time.Now().Unix(), - }) - - _, err := service.Acquire(context.Background(), true) - require.NoError(t, err) - - stored := readLockValue(t, db) - require.Equal(t, service.processID, stored.ProcessID) - require.Equal(t, service.hostID, stored.HostID) - require.Greater(t, stored.ExpiresAt, time.Now().Unix()) - }) - - t.Run("force acquisition returns wait duration when stealing active lock", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - existing := lockValue{ - ProcessID: 99, - HostID: "other-host", - LockID: "other-lock-id", - ExpiresAt: time.Now().Add(ttl).Unix(), - } - insertLock(t, db, existing) - - waitUntil, err := service.Acquire(context.Background(), true) - require.NoError(t, err) - require.WithinDuration(t, time.Unix(existing.ExpiresAt, 0), waitUntil, time.Second) - }) - - t.Run("force acquisition does not wait when lock id is unchanged", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - insertLock(t, db, lockValue{ - ProcessID: 99, - HostID: "other-host", - LockID: service.lockID, - ExpiresAt: time.Now().Add(ttl).Unix(), - }) - - waitUntil, err := service.Acquire(context.Background(), true) - require.NoError(t, err) - require.True(t, waitUntil.IsZero()) - }) - - t.Run("returns error when existing lock value is invalid JSON", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - err := db.Create(&model.KV{Key: lockKey, Value: new("this-is-not-json")}).Error - require.NoError(t, err) - - _, err = service.Acquire(context.Background(), false) - require.ErrorContains(t, err, "decode existing lock value") - }) - - t.Run("returns context deadline exceeded when database is locked", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - tx := lockDatabaseForWrite(t, db) - defer tx.Rollback() - - ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) - defer cancel() - - _, err := service.Acquire(ctx, false) - require.ErrorIs(t, err, context.DeadlineExceeded) - require.ErrorContains(t, err, "begin lock transaction") - }) -} - -func TestAppLockServiceRelease(t *testing.T) { - t.Run("removes owned lock", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - err = service.Release(context.Background()) - require.NoError(t, err) - - var row model.KV - err = db.Take(&row, "key = ?", lockKey).Error - require.ErrorIs(t, err, gorm.ErrRecordNotFound) - }) - - t.Run("ignores lock held by another owner", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - existing := lockValue{ - ProcessID: 2, - HostID: "other-host", - ExpiresAt: time.Now().Add(ttl).Unix(), - } - insertLock(t, db, existing) - - err := service.Release(context.Background()) - require.NoError(t, err) - - stored := readLockValue(t, db) - require.Equal(t, existing, stored) - }) - - t.Run("returns context deadline exceeded when database is locked", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - tx := lockDatabaseForWrite(t, db) - defer tx.Rollback() - - ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) - defer cancel() - - err = service.Release(ctx) - require.ErrorIs(t, err, context.DeadlineExceeded) - require.ErrorContains(t, err, "release lock failed") - }) -} - -func TestAppLockServiceRenew(t *testing.T) { - t.Run("extends expiration when lock is still owned", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - before := readLockValue(t, db) - - err = service.renew(context.Background()) - require.NoError(t, err) - - after := readLockValue(t, db) - require.Equal(t, service.processID, after.ProcessID) - require.Equal(t, service.hostID, after.HostID) - require.GreaterOrEqual(t, after.ExpiresAt, before.ExpiresAt) - }) - - t.Run("returns ErrLockLost when lock is missing", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - err := service.renew(context.Background()) - require.ErrorIs(t, err, ErrLockLost) - }) - - t.Run("returns ErrLockLost when ownership changed", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - // Simulate a different process taking the lock. - newOwner := lockValue{ - ProcessID: 9, - HostID: "stolen-host", - ExpiresAt: time.Now().Add(ttl).Unix(), - } - raw, marshalErr := newOwner.Marshal() - require.NoError(t, marshalErr) - updateErr := db.Model(&model.KV{}). - Where("key = ?", lockKey). - Update("value", raw).Error - require.NoError(t, updateErr) - - err = service.renew(context.Background()) - require.ErrorIs(t, err, ErrLockLost) - }) - - t.Run("returns context deadline exceeded when database is locked", func(t *testing.T) { - db := testutils.NewDatabaseForTest(t) - service := newTestAppLockService(t, db) - - _, err := service.Acquire(context.Background(), false) - require.NoError(t, err) - - tx := lockDatabaseForWrite(t, db) - defer tx.Rollback() - - ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) - defer cancel() - - err = service.renew(ctx) - require.ErrorIs(t, err, context.DeadlineExceeded) - }) -} diff --git a/backend/internal/service/e2etest_service.go b/backend/internal/service/e2etest_service.go index 7f0d6c7c..55af7c0e 100644 --- a/backend/internal/service/e2etest_service.go +++ b/backend/internal/service/e2etest_service.go @@ -50,7 +50,6 @@ type TestService struct { appConfigService *appconfig.AppConfigService ldapService *LdapService fileStorage storage.FileStorage - appLockService *AppLockService externalIdPKey jwk.Key } @@ -64,14 +63,13 @@ const ( e2eEmailVerificationToken = "2FZFSoupBdHyqIL65bWTsgCgHIhxlXup" ) -func NewTestService(db *gorm.DB, actors *local.Host, appConfigService *appconfig.AppConfigService, jwtService *JwtService, ldapService *LdapService, appLockService *AppLockService, fileStorage storage.FileStorage) (*TestService, error) { +func NewTestService(db *gorm.DB, actors *local.Host, appConfigService *appconfig.AppConfigService, jwtService *JwtService, ldapService *LdapService, fileStorage storage.FileStorage) (*TestService, error) { s := &TestService{ db: db, actors: actors, appConfigService: appConfigService, jwtService: jwtService, ldapService: ldapService, - appLockService: appLockService, fileStorage: fileStorage, } err := s.initExternalIdP() @@ -746,11 +744,6 @@ func (s *TestService) ResetAppConfig(ctx context.Context) error { return nil } -func (s *TestService) ResetLock(ctx context.Context) error { - _, err := s.appLockService.Acquire(ctx, true) - return err -} - // SyncLdap triggers an LDAP synchronization func (s *TestService) SyncLdap(ctx context.Context) error { dbConfig, err := s.appConfigService.GetConfig(ctx) diff --git a/backend/internal/service/export_service.go b/backend/internal/service/export_service.go index 578f3daf..c6b0c140 100644 --- a/backend/internal/service/export_service.go +++ b/backend/internal/service/export_service.go @@ -109,14 +109,6 @@ func (s *ExportService) dumpTable(table string, types utils.DBSchemaTableTypes, rowMap[col] = vals[i] } - // Skip the app lock row in the kv table - if table == "kv" { - keyPtr, ok := rowMap["key"].(*string) - if ok && keyPtr != nil && *keyPtr == lockKey { - continue - } - } - out.Tables[table] = append(out.Tables[table], rowMap) }