Bind retries and recover uncertain operations

When to use this guide

Use these tasks when a retried HTTP request might repeat an external effect. Install Echo Pro and choose a stable trusted tenant and subject. Run executable examples with:

go test -run Example ./execution/...

The execution overview explains permits, proof limits and retention. The reference lists journal, HTTP and PostgreSQL settings.

Bind a route and replay a conditional result

Create one MiddlewareConfig per operation. The operation must exist in the journal’s policy map. Identity supplies tenant and subject. ReplayCheck authorizes current access before new work or replay. This complete source file includes the shared configuration and request helper. Its first example executes the handler once across two requests and replays the second response with Idempotent-Replayed: true.

package execution_test

import (
  "context"
  "crypto/sha256"
  "encoding/json"
  "fmt"
  "net/http"
  "net/http/httptest"
  "time"

  "github.com/labstack/echo/v5"

  "echo.labstack.com/pro/execution"
  "echo.labstack.com/pro/quota"
)

func exampleHTTPConfig() execution.MiddlewareConfig {
  j, err := execution.New(execution.JournalConfig{
    Namespace: "demo",
    Store: execution.NewMemoryStore(
      nil,
    ),
    Policies: map[string]execution.Policy{
      "write": {
        SafetyVersion:     "v1",
        ReconcilerVersion: "proof/v1",
        Effects: []execution.Effect{
          {
            Name:           "write",
            Provider:       "service/account/v1",
            RequestTimeout: time.Second,
          },
        },
      },
    },
  })
  if err != nil {
    panic(err)
  }
  return execution.MiddlewareConfig{
    Journal: j, Operation: "write",
    // In production these callbacks use the authenticated principal and the
    // operation's current authorization policy; never a client identity header.
    Identity: func(
      *echo.Context,
    ) (string, string, error) {
      return "tenant", "subject", nil
    },
    ReplayCheck: func(*echo.Context) error { return nil },
  }
}

func exampleHTTPRequest(ctx context.Context, e *echo.Echo) *httptest.ResponseRecorder {
  r := httptest.NewRequestWithContext(ctx, http.MethodPost, "/write", nil)
  r.Header.Set("Idempotency-Key", `"write-42"`)
  w := httptest.NewRecorder()
  e.ServeHTTP(w, r)
  return w
}

func ExampleMiddlewareConfig_ToMiddleware() {
  cfg := exampleHTTPConfig()
  mw, err := cfg.ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  calls := 0
  e.POST("/write", func(c *echo.Context) error {
    calls++
    // A conditional success needs no external send or Intent.
    if err := execution.Classify(c, execution.Completed); err != nil {
      return err
    }
    return c.String(200, "already current")
  }, mw)
  first := exampleHTTPRequest(context.Background(), e)
  second := exampleHTTPRequest(context.Background(), e)
  fmt.Println(first.Code, first.Body.String())
  fmt.Println(second.Code, second.Header().Get("Idempotent-Replayed"), calls)
  // Output:
  // 200 already current
  // 200 true 1
}

func ExampleMiddlewareConfig_Prepare() {
  ctx := context.Background()
  events := make(chan quota.ConsumptionEvent, 8)
  limiter, err := quota.New(quota.Config{
    Store: quota.NewMemoryStore(quota.MemoryConfig{}),
    Plans: []quota.Plan{{ID: "pro", Limits: []quota.Limit{
      quota.PerMonth("requests", 100),
      quota.PerMonth("tokens", 10000).Meter("tokens"),
    }}},
    Hooks: quota.Hooks{
      OnConsumption: func(
        _ context.Context,
        event quota.ConsumptionEvent,
      ) {
        events <- event
      },
    },
  })
  if err != nil {
    panic(err)
  }
  cfg := exampleHTTPConfig()
  cfg.NotExecuted = func(c *echo.Context) {
    // Zero refunds work. Omitted requests keeps this HTTP attempt charged.
    quota.Charge(c, quota.Cost{"tokens": 0})
  }
  prepare, err := cfg.Prepare()
  if err != nil {
    panic(err)
  }
  execute, err := cfg.ToMiddleware()
  if err != nil {
    panic(err)
  }
  admit, err := (quota.MiddlewareConfig{Limiter: limiter, Settle: true,
    Identity: func(
      *echo.Context,
    ) (quota.Identity, error) {
      return quota.Identity{Tenant: "tenant"}, nil
    },
    Cost: func(*echo.Context) (quota.Cost, error) {
      return quota.Cost{
        "requests": 1,
        "tokens":   100,
      }, nil
    },
  }).ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  // Body/auth precede this chain; a validator goes between admit and execute.
  e.POST("/write", func(c *echo.Context) error {
    quota.Charge(c, quota.Cost{"tokens": 60})
    if err := execution.Classify(c, execution.Completed); err != nil {
      return err
    }
    return c.String(200, "done")
  }, prepare, admit, execute)
  first := exampleHTTPRequest(ctx, e)
  second := exampleHTTPRequest(ctx, e)
  // Settlement is asynchronous. Shutdown drains it without a guessed sleep.
  if err := limiter.Close(ctx); err != nil {
    panic(err)
  }
  close(events)
  amounts := quota.Cost{}
  for event := range events {
    amounts[event.Meter] += event.Amount
  }
  fmt.Println(first.Code, second.Code, second.Header().Get("Idempotent-Replayed"))
  fmt.Println("requests:", amounts["requests"], "tokens:", amounts["tokens"])
  // Output:
  // 200 200 true
  // requests: 2 tokens: 60
}

func ExampleAttemptFrom() {
  ctx := context.Background()
  cfg := exampleHTTPConfig()
  mw, err := cfg.ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  e.POST("/write", func(c *echo.Context) error {
    attempt := execution.AttemptFrom(c)
    digest := sha256.Sum256([]byte("durable request 42"))
    start := time.Now() // Anchor the send deadline before the call.
    permit, err := attempt.Intent(
      c.Request().Context(),
      "write",
      "payload/write-42",
      digest,
      execution.Token{1},
    )
    if err != nil {
      return err
    } // A refusal never authorizes a send.
    deadline := start.Add(permit.ValidFor)
    if !time.Now().Before(deadline) {
      return &execution.AmbiguityError{}
    }
    // One identical provider request uses permit.Key and this deadline, with
    // SDK retries/redirects disabled. Here the provider succeeds, then the
    // handler is interrupted before the journal records completion.
    panic("simulated interruption after provider success")
  }, mw)
  func() {
    defer func() {
      if recover() == nil {
        panic("expected interruption")
      }
    }()
    exampleHTTPRequest(ctx, e)
  }()
  w := exampleHTTPRequest(ctx, e)
  var problem execution.Problem
  if err := json.Unmarshal(w.Body.Bytes(), &problem); err != nil {
    panic(err)
  }
  fmt.Println(w.Code, problem.Code, problem.SafeNewKey)
  // A trusted operator authorizes this target tenant before accessing the ref,
  // verifies provider success, and reconstructs its HTTP result contract.
  operator := cfg.Journal.Operator()
  record, err := operator.Get(ctx, problem.OperationRef)
  if err != nil {
    panic(err)
  }
  prepared, err := operator.PrepareResolve(ctx, execution.ResolveRequest{
    Ref:              record.Ref,
    ExpectedRevision: record.Revision,
    ID:               "decision-42",
    Decision:         execution.Applied,
    Actor:            "trusted-operator",
    Reason:           "provider success verified",
    Result: &execution.Result{
      Status: 200,
      Codec:  "http/v1",
      Headers: map[string][]string{
        "Content-Type": {"text/plain"},
      },
      Body: []byte("recovered"),
    },
  })
  if err != nil {
    panic(err)
  }
  if _, err := operator.Resolve(ctx, prepared); err != nil {
    panic(err)
  }
  w = exampleHTTPRequest(ctx, e)
  fmt.Println(w.Code, w.Body.String(), w.Header().Get("Idempotent-Replayed"))
  // Output:
  // 409 operation_uncertain false
  // 200 recovered true
}

func ExampleClassify() {
  ctx := context.Background()
  cfg := exampleHTTPConfig()
  mw, err := cfg.ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  e.POST("/write", func(c *echo.Context) error {
    if err := execution.Classify(c, execution.Failed); err != nil {
      return err
    }
    return echo.NewHTTPError(422, "business rule rejected")
  }, mw)
  first := exampleHTTPRequest(ctx, e)
  second := exampleHTTPRequest(ctx, e)
  fmt.Println(first.Code, second.Code, second.Header().Get("Idempotent-Replayed"))
  // Output: 422 422 true
}

The fixture callbacks use fixed tenant and subject values and permissive replay access. Replace both with application authentication and current resource policy. Configure request fingerprints for every input that changes the operation. Retain the same idempotency key across retries. Do not treat the key as an access token.

Place preparation before quota and validation

Put Prepare() before quota if requests with a malformed key should fail without a quota charge. Keep authentication before preparation. Put validation between quota and execution if failed validation should still count against quota. NotExecuted refunds unused work through quota.Charge while leaving omitted request units charged.

The complete HTTP source’s ExampleMiddlewareConfig_Prepare reserves 100 tokens, charges actual use of 60, and performs a replay. It drains settlement with Limiter.Close. Its output is two successful HTTP attempts, requests: 2 and tokens: 60. Validation that exits before execution must perform its own refund.

Persist intent and recover an interrupted provider success

ExampleAttemptFrom in the same HTTP source persists an intent and simulates interruption after provider success. Its retry returns 409 operation_uncertain with safe_new_key=false. The authorized operator reads the record and verifies provider success. It prepares an applied resolution with a reconstructed HTTP result. The next request replays 200 recovered true.

Replace simulated provider success with one identical provider request under the permit deadline. Reconstruct it from a durable payload reference. Disable hidden SDK retries and redirects. Store the decision ID and the prepared request, so you can retry if the response is lost. Do not switch keys to work around uncertainty.

Freeze problem responses through Echo’s error handler

Install the problem adapter as Echo’s HTTPErrorHandler, keeping its previous handler as fallback. Include Content-Type, Content-Language and Cache-Control in ReplayHeaders. The example freezes a typed application problem response. It changes rendering language and checks that replay keeps the original English bytes:

package execution_test

import (
  "bytes"
  "context"
  "encoding/json"
  "fmt"
  "net/http/httptest"

  "github.com/labstack/echo/v5"

  "echo.labstack.com/pro/execution"
  "echo.labstack.com/pro/problem"
  "echo.labstack.com/pro/problem/echoadapter"
)

// orderRejectedError is an application error; any type with Problem() qualifies.
type orderRejectedError struct{}

func (orderRejectedError) Error() string { return "order rejected" }
func (orderRejectedError) Problem() problem.Details {
  return problem.Details{
    Type:    "https://shop.example/problems/order-rejected",
    Title:   "Order rejected",
    TitleJA: "注文は受け付けられません",
    Status:  422,
    Code:    "order_rejected",
  }
}

// Execution renders a handler's error by calling Echo's HTTPErrorHandler
// inside its private buffer, before classification and freezing. Install the
// problem adapter there, with the previous handler as the explicit fallback;
// problem middleware inside execution cannot render. Replay sends only
// allowlisted stored headers, so list Content-Type, Content-Language and
// Cache-Control. The frozen bytes replay unchanged after a language change.
func ExampleMiddlewareConfig_ToMiddleware_problemAdapter() {
  cfg := exampleHTTPConfig()
  cfg.ReplayHeaders = []string{"Content-Type", "Content-Language", "Cache-Control"}
  mw, err := cfg.ToMiddleware()
  if err != nil {
    panic(err)
  }
  e := echo.New()
  fallback := e.HTTPErrorHandler
  install := func(lang problem.Language) {
    h, err := echoadapter.ErrorHandlerConfig{
      Render: problem.RenderConfig{Language: lang},
    }.ToErrorHandler(fallback)
    if err != nil {
      panic(err)
    }
    e.HTTPErrorHandler = h
  }
  install(problem.English)
  e.POST("/write", func(c *echo.Context) error {
    // A business refusal is classified, then frozen with its bytes.
    if err := execution.Classify(c, execution.Failed); err != nil {
      return err
    }
    return orderRejectedError{}
  }, mw)
  show := func(w *httptest.ResponseRecorder) {
    h := w.Header()
    fmt.Println(w.Code, h.Get("Content-Type"), h.Get("Content-Language"),
      h.Get("Cache-Control"), h.Get("Idempotent-Replayed") == "true")
    var body bytes.Buffer
    if err := json.Indent(&body, w.Body.Bytes(), "", "  "); err != nil {
      panic(err)
    }
    fmt.Print(body.String())
  }
  show(exampleHTTPRequest(context.Background(), e))
  install(problem.Japanese)
  show(exampleHTTPRequest(context.Background(), e))
  // Output:
  // 422 application/problem+json en no-store false
  // {
  //   "type": "https://shop.example/problems/order-rejected",
  //   "title": "Order rejected",
  //   "status": 422,
  //   "code": "order_rejected"
  // }
  // 422 application/problem+json en no-store true
  // {
  //   "type": "https://shop.example/problems/order-rejected",
  //   "title": "Order rejected",
  //   "status": 422,
  //   "code": "order_rejected"
  // }
}

Do not render problems with middleware placed inside execution. Execution buffers the response, and that middleware cannot write to the buffer. Use the configured error handler. Classify a known business refusal with execution.Failed before returning the typed error. An unclassified error after intent remains uncertain.

Instrument provider timing without resending

Capture the local start before obtaining a permit. Pass the permit and start to Attempt.Call. Its callback executes once; retry diagnostic persistence through RecordSendTiming, never by repeating the provider call. The executable timing example is:

package execution_test

import (
  "context"
  "crypto/rand"
  "fmt"
  "time"

  "echo.labstack.com/pro/execution"
)

func ExampleAttempt_Call() {
  ctx := context.Background()
  journal, err := execution.New(execution.JournalConfig{
    Namespace: "example",
    Store: execution.NewMemoryStore(
      nil,
    ),
    Policies: map[string]execution.Policy{
      "dispatch": {
        Scope:             "dispatch",
        SafetyVersion:     "v1",
        ReconcilerVersion: "v1",
        Effects: []execution.Effect{
          {
            Name:           "send",
            Provider:       "example/v1",
            RequestTimeout: time.Second,
          },
        },
      },
    },
  })
  if err != nil {
    fmt.Println(err)
    return
  }
  attempt, _, err := journal.Begin(ctx, execution.Request{
    Operation: "dispatch",
    Key: execution.Key{
      Tenant:  "example",
      Subject: "worker",
      Scope:   "dispatch",
      Value:   "job-1",
    },
  })
  if err != nil {
    fmt.Println(err)
    return
  }
  var id execution.Token
  if _, err = rand.Read(id[:]); err != nil {
    fmt.Println(err)
    return
  }
  admittedAt := time.Now() // capture before the store call, so its latency counts
  permit, err := attempt.Intent(ctx, "send", "owned-payload-reference", [32]byte{1}, id)
  if err != nil {
    fmt.Println(err)
    return
  }
  timing, err := attempt.Call(ctx, "send", permit, admittedAt, func(
    call context.Context,
  ) (string, string, error) {
    // Replace with one provider call using permit.Key and the identical request.
    // This example callback performs no external work and observes cancellation.
    if err := call.Err(); err != nil {
      return "uncertain", "", err
    }
    return "succeeded", "", nil
  })
  if err != nil {
    fmt.Println(err)
    return
  } // retry only timing persistence, never the callback
  fmt.Println(timing.Outcome)
  // Business finalization is separate from diagnostic timing.
  if _, err = attempt.Complete(ctx, execution.Result{Status: 204}); err != nil {
    fmt.Println(err)
  }
  // Output: succeeded
}

Keep OnTransition and OnError prompt and concurrency safe. Do not reenter the same attempt from those callbacks. Set MaxResultBytes, MaxNamespaceBytes and MaxNamespaceRecords for stored data. These are storage limits; they do not cap your process heap.

Use PostgreSQL and local transactions

Apply pgstore.Migrate with a deployment pool and reviewed schema configuration, then pgstore.New with a caller-owned runtime pool. If a lost record must not allow a second send, use synchronous commit and replication. You can read the SQL through pgstore.Migrations. Apply it only with Migrate.

Use TxRunner only for local transactional SQL. Its callback cannot perform provider I/O. This compiled source demonstrates transaction handling and distinguishes a known rollback from an unknown COMMIT:

package pgstore_test

import (
  "context"
  "errors"
  "fmt"
  "time"

  "github.com/jackc/pgx/v5/pgxpool"

  "echo.labstack.com/pro/execution"
  "echo.labstack.com/pro/execution/pgstore"
)

// ExampleTxRunner_RunTx commits an order row with its execution binding and
// stored result in one transaction. A replay returns the stored result
// without running the callback; NotApplied rolls the order back but still
// records a binding whose SafeNewKey is true.
//
// This example is compile-only: it needs a reachable PostgreSQL database
// whose "execution" schema was already created with Migrate, plus the
// application's own items(id, stock) and orders(item) tables.
func ExampleTxRunner_RunTx() {
  ctx := context.Background()
  pool, err := pgxpool.New(ctx, "postgres:///orders?default_query_exec_mode=exec")
  if err != nil {
    return
  }
  defer pool.Close()
  store, err := pgstore.New(ctx, pool, pgstore.StoreConfig{
    Schema:    "execution",
    Namespace: "orders",
  })
  if err != nil {
    return
  }
  runner, err := pgstore.NewTxRunner(store, pgstore.TxRunnerConfig{
    Namespace: "orders",
    Policies: map[string]execution.Policy{
      "create-order": {Mode: execution.TransactionalMode, SafetyVersion: "v1"},
    },
  })
  if err != nil {
    return
  }
  key := execution.Key{Tenant: "tenant-1", Subject: "user-1", Value: "client-key"}
  done, err := runner.RunTx(ctx, pgstore.TxRequest{
    Operation:   "create-order",
    Key:         key,
    Fingerprint: [32]byte{1},
  },
    func(ctx context.Context, tx *pgstore.Tx) (pgstore.TxOutcome, error) {
      var stock int
      if err := tx.QueryRow(
        ctx,
        "SELECT stock FROM items WHERE id=$1 FOR UPDATE",
        "item-1",
      ).Scan(&stock); err != nil {
        return pgstore.TxOutcome{}, err
      }
      if stock < 1 {
        // No protected work remains. The record says no effect applied.
        return pgstore.TxOutcome{
          Decision: execution.NotApplied,
          Result: &execution.Result{
            Status: 409,
            Body: []byte(
              `{"error":"out_of_stock"}`,
            ),
          },
        }, nil
      }
      if _, err := tx.Exec(
        ctx,
        "UPDATE items SET stock=stock-1 WHERE id=$1",
        "item-1",
      ); err != nil {
        return pgstore.TxOutcome{}, err
      }
      if _, err := tx.Exec(
        ctx,
        "INSERT INTO orders(item) VALUES ($1)",
        "item-1",
      ); err != nil {
        return pgstore.TxOutcome{}, err
      }
      return pgstore.TxOutcome{
        Decision: execution.Applied,
        Result: &execution.Result{
          Status: 201,
          Body:   []byte(`{"created":true}`),
        },
      }, nil
    })
  if unknown, ok := errors.AsType[*pgstore.TxCommitUnknownError](
    err,
  ); ok && unknown != nil {
    return // Retry with the identical request; never with a new key.
  }
  if err != nil {
    return
  }
  fmt.Println(done.Answer.Status, done.Answer.Replayed)
}

// Neither a rollback nor an unknown COMMIT authorizes a new key. An unknown
// COMMIT is unavailable: retry the identical request to replay or execute.
func ExampleTxCommitUnknownError() {
  var unknown error = &pgstore.TxCommitUnknownError{}
  var rolledBack error = &pgstore.TxError{
    Phase:             pgstore.TxPhaseCallback,
    Cause:             context.Canceled,
    RollbackConfirmed: true,
  }
  fmt.Println(errors.Is(unknown, execution.ErrUnavailable), unknown)
  fmt.Println(errors.Is(rolledBack, context.Canceled), rolledBack)
  // Output:
  // true execution: transactional commit outcome unknown
  // true execution: transactional callback failed
}

// Configuration is validated before any I/O, and errors are typed for
// inspection with errors.Is and errors.AsType. New validates MigrateTimeout
// even though only Migrate uses it.
func ExampleNewTxRunner() {
  _, err := pgstore.NewTxRunner(nil, pgstore.TxRunnerConfig{Namespace: "orders"})
  if e, ok := errors.AsType[*execution.ConfigError](err); ok {
    fmt.Println("runner:", e.Field)
  }
  ctx := context.Background()
  pool, err := pgxpool.New(
    ctx,
    "postgres://localhost:1/unused",
  ) // Connects lazily; never dialed here.
  if err != nil {
    return
  }
  defer pool.Close()
  _, err = pgstore.New(ctx, pool, pgstore.StoreConfig{MigrateTimeout: 2 * time.Hour})
  if e, ok := errors.AsType[*execution.ConfigError](err); ok {
    fmt.Println("store:", e.Field)
  }
  var rolledBack error = &pgstore.TxError{
    Phase: pgstore.TxPhaseFinalize,
    Cause: execution.ErrUnavailable,
  }
  if e, ok := errors.AsType[*pgstore.TxError](rolledBack); ok {
    fmt.Println(errors.Is(
      rolledBack,
      &pgstore.TxError{},
    ), e.Phase, errors.Is(rolledBack, execution.ErrUnavailable))
  }
  // Output:
  // runner: store
  // store: MigrateTimeout
  // true finalize true
}

The database examples without Output compile during Go tests and require real pools to execute. Test them against your assigned backend before relying on commit behavior. A commit error does not authorize a second external effect or a new operation key.

Run reconciliation with an owned lifecycle

Create one operator and keep it for the scheduler’s lifetime. Map scopes explicitly. Set namespace and target authorization callbacks. Start the scheduler under an application context, cancel it at shutdown and join it. If Run stops with an error, a prepared request may still be pending. Retry it with the original operator before starting replacement work.

package execution_test

import (
  "context"
  "errors"
  "fmt"
  "time"

  "echo.labstack.com/pro/execution"
)

func ExampleNewReconcileScheduler() {
  now := time.Now()
  store := execution.NewMemoryStore(
    func() time.Time { return now },
  ) // Tests inject a clock; production uses the default.
  p := execution.Policy{
    SafetyVersion:     "v1",
    Scope:             "write",
    ReconcilerVersion: "proof/v1",
    Effects: []execution.Effect{
      {
        Name:           "write",
        Provider:       "service/account/v1",
        RequestTimeout: time.Second,
      },
    },
  }
  j, err := execution.New(execution.JournalConfig{
    Namespace: "example",
    Store:     store,
    Policies:  map[string]execution.Policy{"write": p},
  })
  if err != nil {
    panic(err)
  }
  a, _, err := j.Begin(context.Background(), execution.Request{
    Operation: "write",
    Key: execution.Key{
      Tenant:  "tenant",
      Subject: "principal",
      Value:   "key",
    },
  })
  if err != nil {
    panic(err)
  }
  if _, err = a.Intent(
    context.Background(),
    "write",
    "durable/payload",
    [32]byte{1},
    execution.Token{1},
  ); err != nil {
    panic(err)
  }
  r, err := a.Uncertain(
    context.Background(),
    execution.Evidence{},
  ) // The provider reply was lost.
  if err != nil {
    panic(err)
  }
  now = now.Add(6 * time.Minute) // Past the send barrier: LastSendBy + 5 minutes.

  ctx, cancel := context.WithCancel(
    context.Background(),
  ) // Application-owned lifecycle.
  defer cancel()
  // Retain one Operator: prepared handles, including a pending one returned by
  // Run, can only be retried through the Operator instance that prepared them.
  operator := j.Operator()
  passes := 0
  scheduler, err := execution.NewReconcileScheduler(execution.ReconcileSchedulerConfig{
    Operator: operator,
    Scopes: map[string]execution.ScheduledReconciler{"write": {
      Reconciler: func(
        _ context.Context,
        _ *execution.ReconcileClaim,
        _ execution.EffectRecord,
      ) (
        execution.Verdict,
        error,
      ) {
        // Probe using the stored request key. Never rerun business work.
        return execution.Verdict{
          Outcome: execution.EffectApplied,
          Proof: execution.Proof{
            Kind:          execution.ProviderSuccess,
            Version:       "proof/v1",
            Reference:     "redacted/success/1",
            Authoritative: true,
          },
        }, nil
      },
      Completion: func(context.Context, execution.Record, []execution.Verdict) (
        bool,
        *execution.Result,
        error,
      ) {
        return true, nil, nil
      },
    }},
    PollInterval: time.Second,
    AuthorizeNamespace: func(context.Context, string) error {
      passes++
      if passes > 1 {
        cancel() // Stop after one pass, as an application shutdown would.
      }
      return nil
    },
    AuthorizeTarget: func(_ context.Context, target execution.ReconcileTarget) (
      bool,
      error,
    ) {
      return target.Tenant == "tenant", nil // Current per-target authorization.
    },
  })
  if err != nil {
    panic(err)
  }
  err = scheduler.Run(ctx) // Blocks; run it on an application-owned goroutine.
  if stopped, ok := errors.AsType[*execution.SchedulerError](err); ok {
    if handle, held := stopped.Pending(); held {
      // Reauthorize handle.Ref, then retry that handle.
      if _, retryErr := operator.Reconcile(
        context.Background(),
        handle.Prepared,
      ); retryErr != nil {
        panic(retryErr)
      }
    }
  }
  fmt.Println(errors.Is(err, context.Canceled))
  got, err := operator.Get(context.Background(), r.Ref)
  if err != nil {
    panic(err)
  }
  fmt.Println(got.State, got.Proof, got.ReconcileAttempts, got.Receipts[0].ID[:4])
  // Output:
  // true
  // completed proven 1 sch_
}

Expose authorized inspection through executionadmin in the console. Follow the client guidance for operation uncertain, policy changed and result unavailable. If a saved result is missing, the middleware returns operation_result_unavailable. It does not rerun the handler.