Trigger Response Undeliverable

README.md

@@ -24,12 +24,18 @@ func main() {
     Namespace:      "abc",
     PartitionID:    0,
     PartitionTotal: 1,
-    TriggerFn: func(context.Context, *api.TriggerRequest) bool {
+    TriggerFn: func(context.Context, *api.TriggerRequest) *api.TriggerResponse {
       // Do something with your trigger here.
-      // Return true if the trigger was successful, false otherwise.
-      // Note, returning false will cause the job to be retried according to
-      // the Jobs configurable FailurePolicy.
-      return true
+      // Return SUCCESS if the trigger was successful, FAILED if the trigger
+      // failed and should be subject to the FailurePolicy, or UNDELIVERABLE if
+      // the job is currently undeliverable and should be moved to the staging
+      // queue. Use `cron.DeliverablePrefixes` elsewhere to mark jobs with the
+      // given prefixes as now deliverable.
+	  return &api.TriggerResponse{
+        Result: api.TriggerResponseResult_SUCCESS,
+        // Result: api.TriggerResponseResult_FAILED,
+        // Result: api.TriggerResponseResult_UNDELIVERABLE,
+      }
     },
   })
   if err != nil {
@@ -43,11 +49,14 @@ func main() {
   meta, _ := anypb.New(wrapperspb.String("world"))
   tt := time.Now().Add(time.Second).Format(time.RFC3339)
 
-  cron.Add(context.TODO(), "my-job", &api.Job{
+  err = cron.Add(context.TODO(), "my-job", &api.Job{
     DueTime:  &tt,
     Payload:  payload,
     Metadata: meta,
   })
+  if err != nil {
+    panic(err)
+  }
 }
 ```
 
@@ -73,12 +82,19 @@ A Job itself is made up of the following fields:
   Optional.
 - `FailurePolicy` Controls whether the Job should be retired if the trigger
   function returns false. `Drop` doesn't retry the job, `Constant `Constant` will
-  constantly retry the job trigger for a configurable internal, up to a configurable
+  constantly retry the job trigger for a configurable interval, up to a configurable
   maximum number of retries (which could be infinite). By default, Jobs have a
   `Constant` policy, with a 1s interval and 3 maximum retries.
 
 A job must have *at least* either a `Schedule` or a `DueTime` set.
 
+### Undeliverable Jobs
+
+It can be the case that a job trigger hasn't actually _failed_, but instead is simply undeliverable at the current time.
+In such cases, the trigger function can return `UNDELIVERABLE` to indicate that the job should be moved to the "staging queue" to be held until it can be delivered.
+Staged jobs can be marked as deliverable again by calling `cron.DeliverablePrefixes` with the prefixes of those job names.
+Jobs whose name match these prefixes will be re-enqueued for delivery.
+
 ## Leadership
 
 The cron scheduler uses a partition key ownership model to ensure that only one partition instance of the scheduler is running at any given time.

api/api.go

@@ -11,10 +11,10 @@ import (
 
 // TriggerFunction is the type of the function that is called when a job is
 // triggered.
-// Returning true will "tick" the job forward to the next scheduled time.
-// Returning false will cause the job to be re-enqueued and triggered
-// immediately.
-type TriggerFunction func(context.Context, *TriggerRequest) bool
+// The returne TriggerResponse will indicate whether the Job was successfully
+// triggered, the trigger failed, or the Job need to be put into the staging
+// queue.
+type TriggerFunction func(context.Context, *TriggerRequest) *TriggerResponse
 
 // API is the interface for interacting with the cron instance.
 type API interface {
@@ -33,6 +33,17 @@ type API interface {
 
 	// List lists all jobs under a given job name prefix.
 	List(ctx context.Context, prefix string) (*ListResponse, error)
+
+	// DeliverablePrefixes registers the given Job name prefixes as being
+	// deliverable. Any Jobs that reside in the staging queue because they were
+	// undeliverable at the time of trigger but whose names match these prefixes
+	// will be immediately re-triggered.
+	// The returned CancelFunc should be called to unregister the prefixes,
+	// meaning these prefixes are no longer delivable by the caller. Duplicate
+	// Prefixes may be called together and will be pooled together, meaning that
+	// the prefix is still active if there is at least one DeliverablePrefixes
+	// call that has not been unregistered.
+	DeliverablePrefixes(ctx context.Context, prefixes ...string) (context.CancelFunc, error)
 }
 
 // Interface is a cron interface. It schedules and manages job which are stored