cgalo5758
c4bb1ba585
Sweep stranded pending claims at boot and on a Temporal schedule while preserving evidence-based abandonment semantics. Apply occupancy and name-policy checks to carves by operator-root owners without affecting direct operator placements.
74 lines
2.7 KiB
Go
74 lines
2.7 KiB
Go
package domains
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
|
|
"git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/common"
|
|
"go.temporal.io/sdk/workflow"
|
|
)
|
|
|
|
// SweepExpiredClaimsOutput reports one expiry-sweep pass.
|
|
type SweepExpiredClaimsOutput struct {
|
|
Candidates int
|
|
Expired int
|
|
Failed int
|
|
}
|
|
|
|
// SweepExpiredClaimsActivity terminates every pending claim whose verification
|
|
// window has elapsed (claim-expiry-and-carve-guards D1).
|
|
//
|
|
// It is idempotent — each write is guarded on `pending`, so re-running it, or
|
|
// racing a claim's own verification workflow, changes nothing it should not —
|
|
// and it never returns an error for a single unswept row: the counts say what
|
|
// happened and the next pass retries.
|
|
func (a *Activities) SweepExpiredClaimsActivity(ctx context.Context) (SweepExpiredClaimsOutput, error) {
|
|
result, err := a.Registry.SweepExpiredClaims(ctx)
|
|
out := SweepExpiredClaimsOutput{
|
|
Candidates: result.Candidates,
|
|
Expired: result.Expired,
|
|
Failed: result.Failed,
|
|
}
|
|
if err != nil {
|
|
if a.Logger != nil {
|
|
a.Logger.Warn("domain claim expiry sweep reported failures",
|
|
slog.Int("candidates", out.Candidates),
|
|
slog.Int("expired", out.Expired),
|
|
slog.Int("failed", out.Failed),
|
|
slog.Any("error", err))
|
|
}
|
|
// A read failure leaves nothing swept and is worth retrying; per-row
|
|
// failures already counted above are not, and the next scheduled pass
|
|
// picks those rows up regardless.
|
|
if out.Candidates == 0 {
|
|
return out, err
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// SweepExpiredClaimsWorkflow expires pending claims past their deadline. It
|
|
// runs on a recurring schedule as the backstop to the workflow-primary expiry
|
|
// path: a claim's own VerifyClaimWorkflow normally marks it expired at the
|
|
// deadline, but that is the ONLY other trigger, so a worker crash, an operator
|
|
// terminating the execution, or a replay-incompatible deploy would otherwise
|
|
// leave the claim pending forever — holding its whole subtree against every
|
|
// workspace and burning a pending-cap slot.
|
|
//
|
|
// The console also runs the same sweep once per boot, unconditionally, because
|
|
// this schedule exists only where Temporal is configured.
|
|
func SweepExpiredClaimsWorkflow(ctx workflow.Context) error {
|
|
logger := workflow.GetLogger(ctx)
|
|
ctx = workflow.WithActivityOptions(ctx, common.DefaultActivityOptions())
|
|
|
|
var activities *Activities
|
|
var out SweepExpiredClaimsOutput
|
|
if err := workflow.ExecuteActivity(ctx, activities.SweepExpiredClaimsActivity).Get(ctx, &out); err != nil {
|
|
logger.Error("SweepExpiredClaimsActivity failed", "error", err)
|
|
return err
|
|
}
|
|
logger.Info("SweepExpiredClaimsWorkflow completed",
|
|
"candidates", out.Candidates, "expired", out.Expired, "failed", out.Failed)
|
|
return nil
|
|
}
|