Documentation
¶
Index ¶
Constants ¶
This section is empty.
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), ) )
Functions ¶
This section is empty.
Types ¶
type ClaimSweeper ¶
type ClaimSweeper struct {
// contains filtered or unexported fields
}
ClaimSweeper reaps expired upload claims after a grace window. Backend object checks run before the database transaction so remote storage latency never extends claim-row locks. The final transaction uses conditional updates/deletes so concurrent complete_upload calls win or lose cleanly without orphaning objects.
func NewClaimSweeper ¶
func NewClaimSweeper( db orm.DB, service storage.Service, claimStore store.ClaimStore, partStore store.UploadPartStore, 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. Errors at each stage (listing expired rows, per-claim backend probe, transactional cleanup) are logged; the next tick re-reads the expired set, so transient failures self-heal.
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.
Done + FileDeleted publish (on success) and Defer + DeadLetter publish (on terminal failure) each commit in a single transaction via db.RunInTx, so the bookkeeping and the outbox event flip atomically. The transient-retry Defer (intermediate failure) runs in its own RunInTx for ORM consistency but carries no co-committed event. The backend object delete itself runs outside the transaction — it is idempotent (ErrObjectNotFound is silently absorbed), so a retried tick after a transaction-side crash converges without leaking objects or events.
func NewDeleteWorker ¶
func NewDeleteWorker( service storage.Service, deleteQueue store.DeleteQueue, bus event.Bus, db orm.DB, 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.
Source Files