worker

Documentation

Index

• Variables
• type ClaimSweeper
  • func NewClaimSweeper(db orm.DB, claimStore store.ClaimStore, deleteQueue store.DeleteQueue, ...) *ClaimSweeper
  • func (s *ClaimSweeper) Run(ctx context.Context)
• type DeleteWorker
  • func NewDeleteWorker(service storage.Service, deleteQueue store.DeleteQueue, ...) *DeleteWorker
  • func (w *DeleteWorker) Run(ctx context.Context)

Variables

var (

	// Module wires the claim sweeper and delete worker into the fx graph
	// and registers their cron jobs.
	Module = fx.Module(
		"vef:storage:worker",

		fx.Provide(NewClaimSweeper),
		fx.Provide(NewDeleteWorker),
		fx.Invoke(registerJobs),
	)
)

Types

type ClaimSweeper

type ClaimSweeper struct {
	// contains filtered or unexported fields
}

ClaimSweeper reaps expired upload claims by translating each into a PendingDelete row and removing the claim entry — both atomically in one transaction. The actual backend work (abort multipart, delete object) is performed asynchronously by DeleteWorker, which inherits retry/backoff/dead-letter from the queue. ClaimSweeper itself is a pure metadata cleaner; if a single batch fails to commit, no rows are dropped and the next tick re-attempts the same set.

func NewClaimSweeper

func NewClaimSweeper(
	db orm.DB,
	claimStore store.ClaimStore,
	deleteQueue store.DeleteQueue,
	cfg *config.StorageConfig,
) *ClaimSweeper

NewClaimSweeper constructs a ClaimSweeper. db is required to wrap the schedule-and-delete pair in a single transaction.

func (*ClaimSweeper) Run

func (s *ClaimSweeper) Run(ctx context.Context)

Run executes one sweep cycle. Safe to invoke from a cron task. Logs and returns on any error; the next tick will pick up the same expired set.

type DeleteWorker

type DeleteWorker struct {
	// contains filtered or unexported fields
}

DeleteWorker drains sys_storage_pending_delete: for each leased row it optionally aborts a multipart session (UploadID != "" and the backend implements storage.Multipart), deletes the underlying object, and either marks the row done or defers it with exponential backoff. Rows that exceed StorageConfig.DeleteMaxAttempts are parked indefinitely and a dead-letter event is published.

func NewDeleteWorker

func NewDeleteWorker(
	service storage.Service,
	deleteQueue store.DeleteQueue,
	publisher event.Publisher,
	cfg *config.StorageConfig,
) *DeleteWorker

NewDeleteWorker constructs a DeleteWorker. The optional multipart capability is resolved once via a type assertion against the backend; processOne consults the resulting handle instead of probing the backend on every iteration.

func (*DeleteWorker) Run

func (w *DeleteWorker) Run(ctx context.Context)

Run executes one drain cycle. Safe to invoke from a cron task.