Removing time window used by worker to query for exposure batches available for export.

internal/export/config.go

@@ -47,6 +47,8 @@ type Config struct {
  TruncateWindow time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
  MinWindowAge time.Duration `envconfig:"MIN_WINDOW_AGE" default:"2h"`
  TTL time.Duration `envconfig:"CLEANUP_TTL" default:"336h"`
+ // The worker considers export batches closed before the now() minus BatchCloseTimeShift minutes ago.
+ BatchCloseTimeShift time.Duration `envconfig:"BATCH_CLOSE_TIME_SHIFT" default:"5m"`
 }
 
 func (c *Config) BlobstoreConfig() *storage.Config {

internal/export/database/export.go

@@ -401,7 +401,7 @@ func (db *ExportDB) AddExportBatches(ctx context.Context, batches []*model.Expor
 }
 
 // LeaseBatch returns a leased ExportBatch for the worker to process. If no work to do, nil will be returned.
-func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.Time) (*model.ExportBatch, error) {
+func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, batchMaxCloseTime time.Time) (*model.ExportBatch, error) {
  // Lookup a set of candidate batch IDs.
  var openBatchIDs []int64
  err := func() error { // Use a func to allow defer conn.Release() to work.
@@ -427,7 +427,7 @@ func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.
  AND
  end_timestamp < $3
  LIMIT 100
- `, model.ExportBatchOpen, model.ExportBatchPending, now)
+ `, model.ExportBatchOpen, model.ExportBatchPending, batchMaxCloseTime)
  if err != nil {
  return err
  }
@@ -475,7 +475,7 @@ func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.
  return err
  }
 
- if status == model.ExportBatchComplete || (expires != nil && status == model.ExportBatchPending && now.Before(*expires)) {
+ if status == model.ExportBatchComplete || (expires != nil && status == model.ExportBatchPending && batchMaxCloseTime.Before(*expires)) {
  // Something beat us to this batch, it's no longer available.
  return nil
  }
@@ -487,7 +487,7 @@ func (db *ExportDB) LeaseBatch(ctx context.Context, ttl time.Duration, now time.
  status = $1, lease_expires = $2
  WHERE
  batch_id = $3
- `, model.ExportBatchPending, now.Add(ttl), bid)
+ `, model.ExportBatchPending, batchMaxCloseTime.Add(ttl), bid)
  if err != nil {
  return err
  }

internal/export/worker.go

@@ -60,10 +60,10 @@ func (s *Server) WorkerHandler(w http.ResponseWriter, r *http.Request) {
 
  // Only consider batches that closed a few minutes ago to allow the publish
  // windows to close properly.
- minutesAgo := time.Now().Add(-5 * time.Minute)
+ batchMaxCloseTime := time.Now().Add(-s.config.BatchCloseTimeShift * time.Second)
 
  // Check for a batch and obtain a lease for it.
- batch, err := exportDB.LeaseBatch(ctx, s.config.WorkerTimeout, minutesAgo)
+ batch, err := exportDB.LeaseBatch(ctx, s.config.WorkerTimeout, batchMaxCloseTime)
  if err != nil {
  logger.Errorf("Failed to lease batch: %v", err)
  continue