cgalo5758
3727ff31d8
Add an append-only ledger of entitlement set rule changes with per-pool effect rows, a preview-and-commit rule change flow, and an automatic drain that settles deferred recomputations. Rules gain a tier reduction policy, resource keys declare over-limit behavior, and the materializer now lowers limits when a rule stops applying. Add entitlement set rule change ledger and preview flow Add an append-only ledger of entitlement set rule changes with a preview-and-commit operator flow. Rule writes now go through an enclosed `core.commit_rule_change` function that files an act row and one obligation per carrying pool, with a drain workflow settling deferred recomputations. The preview dry-runs the materializer with a rule overlay and renders per-pool buckets, reduction-policy disclosures, and provider over-limit consequences. Materializing transactions take a shared advisory rendezvous that rule changes hold exclusively, enforced by a possession assertion. Add History and Entitlement changes surfaces, a rule-less warning on five product-selection surfaces, and a `tier_reduction_policy` column that gates FedWiki parking.
103 lines
4.3 KiB
Go
103 lines
4.3 KiB
Go
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
|
|
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
|
|
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"net/http"
|
|
|
|
"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
|
|
wfEnt "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/entitlements"
|
|
"git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/sdk/client"
|
|
)
|
|
|
|
// retryFailedObligationsCap bounds the rows Retry reads to find which
|
|
// changes to redispatch; the requeue itself is one statement over every
|
|
// failed row of the set, so the cap costs no correctness.
|
|
const retryFailedObligationsCap = 500
|
|
|
|
// startEntitlementRecomputeDrain starts the change's drain workflow the
|
|
// moment its transaction has committed, so the pools a deferred commit owes
|
|
// settle in milliseconds rather than at the poller's next tick. The workflow
|
|
// id is one per change and the conflict policy is USE_EXISTING, so a redispatch
|
|
// lands on the running execution.
|
|
//
|
|
// Nothing here is load-bearing for correctness: the obligations are durable
|
|
// rows, and PollEntitlementRecompute dispatches every change that still owes
|
|
// a pool. A missing Temporal client and a failed start are therefore both
|
|
// logged and passed over, and the commit's own toast stands.
|
|
func (h *OperatorPartialsHandler) startEntitlementRecomputeDrain(ctx context.Context, result *entitlements.CommitResult) {
|
|
if result == nil || result.ChangeID == "" {
|
|
return
|
|
}
|
|
if result.SyncPath != "deferred" && result.Failed == 0 && result.Recomputed >= result.PoolCount {
|
|
return
|
|
}
|
|
if h.TemporalClient == nil {
|
|
h.Logger.Warn("no Temporal client configured; the entitlement recompute drain was not started",
|
|
slog.String("change_id", result.ChangeID))
|
|
return
|
|
}
|
|
if _, err := h.TemporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
|
ID: wfEnt.RecomputeDrainWorkflowID(result.ChangeID),
|
|
TaskQueue: queues.Main,
|
|
WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
|
|
}, wfEnt.EntitlementRecomputeDrainWorkflow, wfEnt.EntitlementRecomputeDrainInput{
|
|
ChangeID: result.ChangeID,
|
|
}); err != nil {
|
|
h.Logger.Error("failed to start the entitlement recompute drain; the poller will pick the change up",
|
|
slog.Any("error", err), slog.String("change_id", result.ChangeID))
|
|
}
|
|
}
|
|
|
|
// RetryEntitlementRecompute handles
|
|
// POST /partials/operator/entitlement-sets/{setID}/recompute/retry — the one
|
|
// operator-caused transition in the obligations' status domain. It returns
|
|
// this set's failed rows to pending with their attempts reset and
|
|
// redispatches each affected change's drain, which is why the changes are
|
|
// read before the requeue: a requeued row is no longer failed.
|
|
//
|
|
// It acts on no other status, so retrying cannot disturb a pool that is
|
|
// still draining or one that has already settled
|
|
// (entitlement-set-history "The one operator-caused transition").
|
|
func (h *OperatorPartialsHandler) RetryEntitlementRecompute(w http.ResponseWriter, r *http.Request) {
|
|
setID, ok := h.pathUUID(w, r, "setID")
|
|
if !ok {
|
|
return
|
|
}
|
|
ctx := r.Context()
|
|
|
|
failed, err := h.EntitlementsQ.ListFailedObligations(ctx, entitlements.ListFailedObligationsParams{
|
|
SetID: setID,
|
|
Limit: retryFailedObligationsCap,
|
|
})
|
|
if err != nil {
|
|
h.Logger.Error("failed to list failed obligations", slog.Any("error", err), slog.String("set_id", setID))
|
|
h.renderEntitlementSetDetailBody(w, r, setID, "", "Failed to retry the recomputation. Details are in the server logs.")
|
|
return
|
|
}
|
|
if err := h.EntitlementsQ.RequeueFailedObligations(ctx, setID); err != nil {
|
|
h.Logger.Error("failed to requeue failed obligations", slog.Any("error", err), slog.String("set_id", setID))
|
|
h.renderEntitlementSetDetailBody(w, r, setID, "", "Failed to retry the recomputation. Details are in the server logs.")
|
|
return
|
|
}
|
|
|
|
seen := make(map[string]bool, len(failed))
|
|
for _, row := range failed {
|
|
if seen[row.ChangeID] {
|
|
continue
|
|
}
|
|
seen[row.ChangeID] = true
|
|
// SyncPath deferred so the helper dispatches rather than
|
|
// short-circuiting: the requeued rows are exactly the work the
|
|
// drain has left to do.
|
|
h.startEntitlementRecomputeDrain(ctx, &entitlements.CommitResult{ChangeID: row.ChangeID, SyncPath: "deferred"})
|
|
}
|
|
|
|
h.renderEntitlementSetDetailBody(w, r, setID, "", "")
|
|
}
|