mirrors/pocket-id
1
0
Fork 0
mirror of https://github.com/pocket-id/pocket-id.git synced 2026-03-24 13:05:06 +00:00
Code Activity

Ensure cleanup jobs have some jitter

Browse Source
This commit is contained in:
ItalyPaleAle
2026-03-04 20:08:57 -08:00
parent 6bc304dd7f
commit 6615ad64a4
6 changed files with 49 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
backend/internal/job/db_cleanup_job.go
View File

@@ -7,7 +7,6 @@ import (
"log/slog" "log/slog"
"time" "time"
"github.com/go-co-op/gocron/v2"
"gorm.io/gorm" "gorm.io/gorm"
"github.com/pocket-id/pocket-id/backend/internal/common" "github.com/pocket-id/pocket-id/backend/internal/common"
@@ -18,17 +17,15 @@ import (
func (s *Scheduler) RegisterDbCleanupJobs(ctx context.Context, db *gorm.DB) error { func (s *Scheduler) RegisterDbCleanupJobs(ctx context.Context, db *gorm.DB) error {
jobs := &DbCleanupJobs{db: db} jobs := &DbCleanupJobs{db: db}
// Run every 24 hours (but with some jitter so they don't run at the exact same time), and now
def := gocron.DurationRandomJob(24*time.Hour-2*time.Minute, 24*time.Hour+2*time.Minute)
return errors.Join( return errors.Join(
s.RegisterJob(ctx, "ClearWebauthnSessions", def, jobs.clearWebauthnSessions, true), s.RegisterJob(ctx, "ClearWebauthnSessions", jobDefWithJitter(24*time.Hour), jobs.clearWebauthnSessions, true),
s.RegisterJob(ctx, "ClearOneTimeAccessTokens", def, jobs.clearOneTimeAccessTokens, true), s.RegisterJob(ctx, "ClearOneTimeAccessTokens", jobDefWithJitter(24*time.Hour), jobs.clearOneTimeAccessTokens, true),
s.RegisterJob(ctx, "ClearSignupTokens", def, jobs.clearSignupTokens, true), s.RegisterJob(ctx, "ClearSignupTokens", jobDefWithJitter(24*time.Hour), jobs.clearSignupTokens, true),
s.RegisterJob(ctx, "ClearEmailVerificationTokens", def, jobs.clearEmailVerificationTokens, true), s.RegisterJob(ctx, "ClearEmailVerificationTokens", jobDefWithJitter(24*time.Hour), jobs.clearEmailVerificationTokens, true),
s.RegisterJob(ctx, "ClearOidcAuthorizationCodes", def, jobs.clearOidcAuthorizationCodes, true), s.RegisterJob(ctx, "ClearOidcAuthorizationCodes", jobDefWithJitter(24*time.Hour), jobs.clearOidcAuthorizationCodes, true),
s.RegisterJob(ctx, "ClearOidcRefreshTokens", def, jobs.clearOidcRefreshTokens, true), s.RegisterJob(ctx, "ClearOidcRefreshTokens", jobDefWithJitter(24*time.Hour), jobs.clearOidcRefreshTokens, true),
s.RegisterJob(ctx, "ClearReauthenticationTokens", def, jobs.clearReauthenticationTokens, true), s.RegisterJob(ctx, "ClearReauthenticationTokens", jobDefWithJitter(24*time.Hour), jobs.clearReauthenticationTokens, true),
s.RegisterJob(ctx, "ClearAuditLogs", def, jobs.clearAuditLogs, true), s.RegisterJob(ctx, "ClearAuditLogs", jobDefWithJitter(24*time.Hour), jobs.clearAuditLogs, true),
) )
} }

19
backend/internal/job/file_cleanup_job.go
View File

@@ -19,14 +19,19 @@ import (
func (s *Scheduler) RegisterFileCleanupJobs(ctx context.Context, db *gorm.DB, fileStorage storage.FileStorage) error { func (s *Scheduler) RegisterFileCleanupJobs(ctx context.Context, db *gorm.DB, fileStorage storage.FileStorage) error {
jobs := &FileCleanupJobs{db: db, fileStorage: fileStorage} jobs := &FileCleanupJobs{db: db, fileStorage: fileStorage}
err := s.RegisterJob(ctx, "ClearUnusedDefaultProfilePictures", gocron.DurationJob(24*time.Hour), jobs.clearUnusedDefaultProfilePictures, false) var errs []error
errs = append(errs,
s.RegisterJob(ctx, "ClearUnusedDefaultProfilePictures", gocron.DurationJob(24*time.Hour), jobs.clearUnusedDefaultProfilePictures, false),
)
// Only necessary for file system storage // Only necessary for file system storage
if fileStorage.Type() == storage.TypeFileSystem { if fileStorage.Type() == storage.TypeFileSystem {
err = errors.Join(err, s.RegisterJob(ctx, "ClearOrphanedTempFiles", gocron.DurationJob(12*time.Hour), jobs.clearOrphanedTempFiles, true)) errs = append(errs,
s.RegisterJob(ctx, "ClearOrphanedTempFiles", gocron.DurationJob(12*time.Hour), jobs.clearOrphanedTempFiles, true),
)
} }
return err return errors.Join(errs...)
} }
type FileCleanupJobs struct { type FileCleanupJobs struct {
@@ -68,7 +73,8 @@ func (j *FileCleanupJobs) clearUnusedDefaultProfilePictures(ctx context.Context)
// If these initials aren't used by any user, delete the file // If these initials aren't used by any user, delete the file
if _, ok := initialsInUse[initials]; !ok { if _, ok := initialsInUse[initials]; !ok {
filePath := path.Join(defaultPicturesDir, filename) filePath := path.Join(defaultPicturesDir, filename)
if err := j.fileStorage.Delete(ctx, filePath); err != nil { err = j.fileStorage.Delete(ctx, filePath)
if err != nil {
slog.ErrorContext(ctx, "Failed to delete unused default profile picture", slog.String("path", filePath), slog.Any("error", err)) slog.ErrorContext(ctx, "Failed to delete unused default profile picture", slog.String("path", filePath), slog.Any("error", err))
} else { } else {
filesDeleted++ filesDeleted++
@@ -95,8 +101,9 @@ func (j *FileCleanupJobs) clearOrphanedTempFiles(ctx context.Context) error {
return nil return nil
} }
if err := j.fileStorage.Delete(ctx, p.Path); err != nil { rErr := j.fileStorage.Delete(ctx, p.Path)
slog.ErrorContext(ctx, "Failed to delete temp file", slog.String("path", p.Path), slog.Any("error", err)) if rErr != nil {
slog.ErrorContext(ctx, "Failed to delete temp file", slog.String("path", p.Path), slog.Any("error", rErr))
return nil return nil
} }
deleted++ deleted++

6
backend/internal/job/ldap_job.go
View File

@@ -4,8 +4,6 @@ import (
"context" "context"
"time" "time"
"github.com/go-co-op/gocron/v2"
"github.com/pocket-id/pocket-id/backend/internal/service" "github.com/pocket-id/pocket-id/backend/internal/service"
) )
@@ -17,8 +15,8 @@ type LdapJobs struct {
func (s *Scheduler) RegisterLdapJobs(ctx context.Context, ldapService *service.LdapService, appConfigService *service.AppConfigService) error { func (s *Scheduler) RegisterLdapJobs(ctx context.Context, ldapService *service.LdapService, appConfigService *service.AppConfigService) error {
jobs := &LdapJobs{ldapService: ldapService, appConfigService: appConfigService} jobs := &LdapJobs{ldapService: ldapService, appConfigService: appConfigService}
// Register the job to run every hour // Register the job to run every hour (with some jitter)
return s.RegisterJob(ctx, "SyncLdap", gocron.DurationJob(time.Hour), jobs.syncLdap, true) return s.RegisterJob(ctx, "SyncLdap", jobDefWithJitter(time.Hour), jobs.syncLdap, true)
} }
func (j *LdapJobs) syncLdap(ctx context.Context) error { func (j *LdapJobs) syncLdap(ctx context.Context) error {

13
backend/internal/job/scheduler.go
View File

@@ -5,6 +5,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"log/slog" "log/slog"
"time"
"github.com/go-co-op/gocron/v2" "github.com/go-co-op/gocron/v2"
"github.com/google/uuid" "github.com/google/uuid"
@@ -33,16 +34,12 @@ func (s *Scheduler) RemoveJob(name string) error {
if job.Name() == name { if job.Name() == name {
err := s.scheduler.RemoveJob(job.ID()) err := s.scheduler.RemoveJob(job.ID())
if err != nil { if err != nil {
errs = append(errs, fmt.Errorf("failed to unqueue job %q with ID %q: %w", name, job.ID().String(), err)) errs = append(errs, fmt.Errorf("failed to dequeue job %q with ID %q: %w", name, job.ID().String(), err))
} }
} }
} }
if len(errs) > 0 {
return errors.Join(errs...) return errors.Join(errs...)
}
return nil
} }
// Run the scheduler. // Run the scheduler.
@@ -105,3 +102,9 @@ func (s *Scheduler) RegisterJob(ctx context.Context, name string, def gocron.Job
return nil return nil
} }
func jobDefWithJitter(interval time.Duration) gocron.JobDefinition {
const jitter = 5 * time.Minute
return gocron.DurationRandomJob(interval-jitter, interval+jitter)
}

4
backend/internal/job/scim_job.go
View File

@@ -16,8 +16,8 @@ type ScimJobs struct {
func (s *Scheduler) RegisterScimJobs(ctx context.Context, scimService *service.ScimService) error { func (s *Scheduler) RegisterScimJobs(ctx context.Context, scimService *service.ScimService) error {
jobs := &ScimJobs{scimService: scimService} jobs := &ScimJobs{scimService: scimService}
// Register the job to run every hour // Register the job to run every hour (with some jitter)
return s.RegisterJob(ctx, "SyncScim", gocron.DurationJob(time.Hour), jobs.SyncScim, true) return s.RegisterJob(ctx, "SyncScim", gocron.DurationJob(24*time.Hour), jobs.SyncScim, true)
} }
func (j *ScimJobs) SyncScim(ctx context.Context) error { func (j *ScimJobs) SyncScim(ctx context.Context) error {

33
backend/internal/service/scim_service.go
View File

@@ -168,7 +168,8 @@ func (s *ScimService) SyncAll(ctx context.Context) error {
errs = append(errs, ctx.Err()) errs = append(errs, ctx.Err())
break break
} }
if err := s.SyncServiceProvider(ctx, provider.ID); err != nil { err = s.SyncServiceProvider(ctx, provider.ID)
if err != nil {
errs = append(errs, fmt.Errorf("failed to sync SCIM provider %s: %w", provider.ID, err)) errs = append(errs, fmt.Errorf("failed to sync SCIM provider %s: %w", provider.ID, err))
} }
} }
@@ -210,26 +211,20 @@ func (s *ScimService) SyncServiceProvider(ctx context.Context, serviceProviderID
} }
var errs []error var errs []error
var userStats scimSyncStats
var groupStats scimSyncStats
// Sync users first, so that groups can reference them // Sync users first, so that groups can reference them
if stats, err := s.syncUsers(ctx, provider, users, &userResources); err != nil { userStats, err := s.syncUsers(ctx, provider, users, &userResources)
errs = append(errs, err) if err != nil {
userStats = stats errs = append(errs, err)
} else { }
userStats = stats
} groupStats, err := s.syncGroups(ctx, provider, groups, groupResources.Resources, userResources.Resources)
stats, err := s.syncGroups(ctx, provider, groups, groupResources.Resources, userResources.Resources)
if err != nil { if err != nil {
errs = append(errs, err) errs = append(errs, err)
groupStats = stats
} else {
groupStats = stats
} }
if len(errs) > 0 { if len(errs) > 0 {
err = errors.Join(errs...)
slog.WarnContext(ctx, "SCIM sync completed with errors", slog.WarnContext(ctx, "SCIM sync completed with errors",
slog.String("provider_id", provider.ID), slog.String("provider_id", provider.ID),
slog.Int("error_count", len(errs)), slog.Int("error_count", len(errs)),
@@ -240,12 +235,14 @@ func (s *ScimService) SyncServiceProvider(ctx context.Context, serviceProviderID
slog.Int("groups_updated", groupStats.Updated), slog.Int("groups_updated", groupStats.Updated),
slog.Int("groups_deleted", groupStats.Deleted), slog.Int("groups_deleted", groupStats.Deleted),
slog.Duration("duration", time.Since(start)), slog.Duration("duration", time.Since(start)),
slog.Any("error", err),
) )
return errors.Join(errs...) return err
} }
provider.LastSyncedAt = new(datatype.DateTime(time.Now())) provider.LastSyncedAt = new(datatype.DateTime(time.Now()))
if err := s.db.WithContext(ctx).Save(&provider).Error; err != nil { err = s.db.WithContext(ctx).Save(&provider).Error
if err != nil {
return err return err
} }
@@ -273,7 +270,7 @@ func (s *ScimService) syncUsers(
// Update or create users // Update or create users
for _, u := range users { for _, u := range users {
existing := getResourceByExternalID[dto.ScimUser](u.ID, resourceList.Resources) existing := getResourceByExternalID(u.ID, resourceList.Resources)
action, created, err := s.syncUser(ctx, provider, u, existing) action, created, err := s.syncUser(ctx, provider, u, existing)
if created != nil && existing == nil { if created != nil && existing == nil {
@@ -434,7 +431,7 @@ func (s *ScimService) syncGroup(
// Prepare group members // Prepare group members
members := make([]dto.ScimGroupMember, len(group.Users)) members := make([]dto.ScimGroupMember, len(group.Users))
for i, user := range group.Users { for i, user := range group.Users {
userResource := getResourceByExternalID[dto.ScimUser](user.ID, userResources) userResource := getResourceByExternalID(user.ID, userResources)
if userResource == nil { if userResource == nil {
// Groups depend on user IDs already being provisioned // Groups depend on user IDs already being provisioned
return scimActionNone, fmt.Errorf("cannot sync group %s: user %s is not provisioned in SCIM provider", group.ID, user.ID) return scimActionNone, fmt.Errorf("cannot sync group %s: user %s is not provisioned in SCIM provider", group.ID, user.ID)