Bound backend calls and recover crash-stranded rows

Every Kubernetes REST config (Harvester, KubeOVN and its derived
OpenBao/DBaaS/endpoints consumers, the admission webhook) now carries
a 30-second request timeout, matching the Rancher client, so a hung
backend releases its goroutine instead of holding it forever. The
KubeOVN startup probe gets a deadline for the same reason. Each
reconcile is bounded per resource so one stuck backend cannot stall
every other PENDING/DELETING resource in the sequential loop.

A create interrupted before the backend confirmed the resource left
a PENDING row with no backend_uid: invisible to ListPending, stuck
forever, its unique name blocked. The reconciler now sweeps such
rows after 20 minutes — strictly above the largest provision budget
(15 minutes for cluster creates, 10 for everything else). Stranded
creates become FAILED with an explanatory message; stranded deletes
are completed by removing the row, since nothing was ever created on
the backend. Every reaped row emits a normal audit event.

Handlers' fire-and-forget provisioning goroutines are now tracked by
a small async.Group, and shutdown drains them for up to 60 seconds
after the HTTP server stops, so a rolling deploy no longer strands
creates mid-flight. The tracker tolerates tasks launched while the
drain is in progress (a raw WaitGroup would panic if srv.Shutdown
timed out with requests still active).

Harvester GetVM and Rancher GetCluster now return a typed not-found
error implementing the NotFound() sentinel the reconciler already
looks for, replacing substring matching on those paths.

Author: HiranAdikari

.github/workflows/authz.yaml

@@ -10,9 +10,9 @@ name: authz-tests
 # real Harvester / KubeOVN cluster.
 #
 # Scope: the role × verb matrix, membership, service-account, tenant,
-# phase-6a slug-recycle, and project-capacity suites. Resource tests
-# (vnet/subnet/vm/cluster/...) genuinely provision on the cluster and are
-# deliberately NOT in scope here.
+# phase-6a slug-recycle, project-capacity, and orphan-reaper (pure-DB crash
+# recovery) suites. Resource tests (vnet/subnet/vm/cluster/...) genuinely
+# provision on the cluster and are deliberately NOT in scope here.
 
 on:
   pull_request:
@@ -48,5 +48,5 @@ jobs:
         # single shared Postgres container + the ~35 status-code assertions.
         run: |
           go test -tags integration -timeout 12m -v \
-            -run 'TestRBAC_|TestMembers_|TestServiceAccountAPI_|TestTenants_|TestPhase6a|TestProjectCap_' \
+            -run 'TestRBAC_|TestMembers_|TestServiceAccountAPI_|TestTenants_|TestPhase6a|TestProjectCap_|TestReaper_' \
             ./test/integration/...

dc-api/cmd/dc-api/main.go

@@ -26,6 +26,7 @@ import (
 	"github.com/wso2/dc-api/internal/api/auth"
 	"github.com/wso2/dc-api/internal/api/handlers"
 	"github.com/wso2/dc-api/internal/api/middleware"
+	"github.com/wso2/dc-api/internal/async"
 	"github.com/wso2/dc-api/internal/config"
 	"github.com/wso2/dc-api/internal/db"
 	"github.com/wso2/dc-api/internal/directory"
@@ -40,6 +41,14 @@ import (
 	"github.com/wso2/dc-api/internal/reconciler"
 )
 
+// provisionDrainTimeout bounds the post-Shutdown wait for the handlers'
+// fire-and-forget provisioning goroutines (async.Group). Long enough for the
+// typical backend submit to finish; deliberately shorter than a Kubernetes
+// terminationGracePeriod escalation would tolerate. Tasks still running when
+// it expires are abandoned and later reaped as FAILED by the reconciler's
+// orphan sweep.
+const provisionDrainTimeout = 60 * time.Second
+
 func main() {
 	// ── Configuration ─────────────────────────────────────────────────────────
 	// config.Load() reads all DCAPI_* environment variables.
@@ -451,6 +460,12 @@ func main() {
 		log.Info().Msg("IdP directory disabled (DCAPI_IDP_* unset) — invite by user_sub only")
 	}
 
+	// ── Async provisioning task group ────────────────────────────────────────
+	// Tracks the fire-and-forget provisioning goroutines the handlers launch
+	// (VM/cluster/network creates + deletes) so shutdown can drain them below
+	// instead of killing a provision mid-flight on a rolling deploy.
+	tasks := &async.Group{}
+
 	// ── Router ────────────────────────────────────────────────────────────────
 	// All wiring happens in NewRouter. main.go does not know about individual routes.
 	router := api.NewRouter(api.RouterDeps{
@@ -492,6 +507,7 @@ func main() {
 		LocalZone:        cfg.LocalZone,
 		AgentRouteReads:  cfg.AgentRouteReads,
 		AgentRouteWrites: cfg.AgentRouteWrites,
+		Tasks:            tasks,
 		Log:              log.Logger,
 	})
 
@@ -525,6 +541,18 @@ func main() {
 	if err := srv.Shutdown(shutdownCtx); err != nil {
 		log.Error().Err(err).Msg("graceful shutdown timed out")
 	}
+
+	// Drain the detached provisioning goroutines (bounded). Without this, a
+	// rolling deploy kills provisions mid-flight, stranding PENDING rows with
+	// no backend_uid — the exact orphans the reconciler sweep exists to reap.
+	log.Info().Dur("timeout", provisionDrainTimeout).Msg("waiting for in-flight provisioning tasks to finish")
+	drainCtx, drainCancel := context.WithTimeout(context.Background(), provisionDrainTimeout)
+	defer drainCancel()
+	if tasks.Wait(drainCtx) {
+		log.Info().Msg("in-flight provisioning tasks drained")
+	} else {
+		log.Warn().Msg("drain timed out — abandoning in-flight provisioning tasks; interrupted creates will be reaped as FAILED by the reconciler's orphan sweep")
+	}
 	log.Info().Msg("DC-API stopped cleanly")
 }
 

dc-api/cmd/webhook/main.go

@@ -65,6 +65,12 @@ type config struct {
 	LogLevel   string `envconfig:"LOG_LEVEL"   default:"info"`
 }
 
+// k8sRequestTimeout bounds every request made through the webhook's
+// rest.Config (both the in-cluster and kubeconfig paths) so a hung API server
+// can't block the NAD-lookup goroutines forever. Mirrors dc-api's provider
+// clients (30s, same as the Rancher http.Client timeout).
+const k8sRequestTimeout = 30 * time.Second
+
 func main() {
 	// ── Config ────────────────────────────────────────────────────────────────
 	var cfg config
@@ -103,6 +109,9 @@ func main() {
 			log.Fatal().Err(err).Msg("webhook: parse kubeconfig failed")
 		}
 	}
+	// Applies to BOTH paths (in-cluster and kubeconfig) — set before the
+	// dynamic client is built from this config.
+	restCfg.Timeout = k8sRequestTimeout
 
 	dynClient, err := dynamic.NewForConfig(restCfg)
 	if err != nil {

dc-api/internal/api/handlers/bastion.go

@@ -13,6 +13,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/wso2/dc-api/internal/api/middleware"
+	"github.com/wso2/dc-api/internal/async"
 	"github.com/wso2/dc-api/internal/db"
 	"github.com/wso2/dc-api/internal/models"
 	"github.com/wso2/dc-api/internal/placement"
@@ -43,7 +44,10 @@ type BastionHandler struct {
 	bastionImage    string // DCAPI_BASTION_IMAGE
 	bastionMgmtNAD  string // DCAPI_BASTION_MGMT_NAD
 	dnsSearchDomain string // DCAPI_VPC_DNS_SEARCH_DOMAIN — same as VMHandler
-	log             zerolog.Logger
+	// tasks tracks the async provisioning goroutines so shutdown can drain
+	// them (bounded). May be nil (tests) — async.Group is nil-receiver-safe.
+	tasks *async.Group
+	log   zerolog.Logger
 }
 
 // NewBastionHandler creates a BastionHandler with injected dependencies.
@@ -52,6 +56,7 @@ func NewBastionHandler(
 	resolve providers.Resolver,
 	defaultRegion, defaultZone string,
 	bastionImage, bastionMgmtNAD, dnsSearchDomain string,
+	tasks *async.Group,
 	log zerolog.Logger,
 ) *BastionHandler {
 	return &BastionHandler{
@@ -62,6 +67,7 @@ func NewBastionHandler(
 		bastionImage:    bastionImage,
 		bastionMgmtNAD:  bastionMgmtNAD,
 		dnsSearchDomain: dnsSearchDomain,
+		tasks:           tasks,
 		log:             log,
 	}
 }
@@ -318,7 +324,7 @@ func (h *BastionHandler) Create(w http.ResponseWriter, r *http.Request) {
 		DNSSearchDomain:  h.dnsSearchDomain,
 		MgmtNAD:          h.bastionMgmtNAD,
 	}
-	go h.asyncProvision(compute, resource.ID, tenantID, projectID, userID, spec, req.Description)
+	h.tasks.Go(func() { h.asyncProvision(compute, resource.ID, tenantID, projectID, userID, spec, req.Description) })
 
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusAccepted)
@@ -423,7 +429,7 @@ func (h *BastionHandler) Delete(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	go func() {
+	h.tasks.Go(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 		defer cancel()
 		if resource.BackendUID == "" {
@@ -434,7 +440,7 @@ func (h *BastionHandler) Delete(w http.ResponseWriter, r *http.Request) {
 			h.log.Error().Err(err).Str("backend_uid", resource.BackendUID).Msg("harvester delete bastion VM failed")
 			_ = h.repo.UpdateStatus(ctx, id, models.StatusFailed, "deletion failed: "+err.Error(), "")
 		}
-	}()
+	})
 
 	w.WriteHeader(http.StatusAccepted)
 }

dc-api/internal/api/handlers/cluster.go

@@ -30,6 +30,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/wso2/dc-api/internal/api/middleware"
+	"github.com/wso2/dc-api/internal/async"
 	"github.com/wso2/dc-api/internal/db"
 	"github.com/wso2/dc-api/internal/models"
 	"github.com/wso2/dc-api/internal/placement"
@@ -49,12 +50,15 @@ type ClusterHandler struct {
 	resolve       providers.Resolver
 	defaultRegion string
 	defaultZone   string
-	log           zerolog.Logger
+	// tasks tracks the async provisioning goroutines so shutdown can drain
+	// them (bounded). May be nil (tests) — async.Group is nil-receiver-safe.
+	tasks *async.Group
+	log   zerolog.Logger
 }
 
 // NewClusterHandler creates a ClusterHandler with injected dependencies.
-func NewClusterHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, log zerolog.Logger) *ClusterHandler {
-	return &ClusterHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, log: log}
+func NewClusterHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, tasks *async.Group, log zerolog.Logger) *ClusterHandler {
+	return &ClusterHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, tasks: tasks, log: log}
 }
 
 // cluster resolves the ClusterProvider for a cluster's (region, zone). Empty
@@ -546,7 +550,7 @@ func (h *ClusterHandler) Create(w http.ResponseWriter, r *http.Request) {
 		},
 		WorkerPools: workerPools,
 	}
-	go h.asyncProvision(cluster, resource.ID, tenantID, projectID, userID, spec)
+	h.tasks.Go(func() { h.asyncProvision(cluster, resource.ID, tenantID, projectID, userID, spec) })
 
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusAccepted)
@@ -708,14 +712,14 @@ func (h *ClusterHandler) Delete(w http.ResponseWriter, r *http.Request) {
 
 	_ = h.repo.UpdateStatus(r.Context(), id, models.StatusDeleting, "deletion requested", "")
 
-	go func() {
+	h.tasks.Go(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 		defer cancel()
 		if err := cluster.DeleteCluster(ctx, resource.BackendUID); err != nil {
 			h.log.Error().Err(err).Str("cluster", resource.Name).Msg("delete cluster failed")
 			_ = h.repo.UpdateStatus(ctx, id, models.StatusFailed, "deletion failed: "+err.Error(), "")
 		}
-	}()
+	})
 
 	w.WriteHeader(http.StatusAccepted)
 }
@@ -852,7 +856,9 @@ func (h *ClusterHandler) AddNodePool(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	go h.asyncAddPool(cluster, clusterID, clusterName, pool, mgmtNAD, tenantSubnetNAD, vmNamespace, req.ImageName)
+	h.tasks.Go(func() {
+		h.asyncAddPool(cluster, clusterID, clusterName, pool, mgmtNAD, tenantSubnetNAD, vmNamespace, req.ImageName)
+	})
 
 	writeJSON(w, http.StatusAccepted, poolToResponse(pool))
 }
@@ -1006,7 +1012,9 @@ func (h *ClusterHandler) ScaleOrUpdateNodePool(w http.ResponseWriter, r *http.Re
 		writeError(w, http.StatusBadGateway, "cannot reach the cluster's zone: "+err.Error())
 		return
 	}
-	go h.asyncPatchPool(cluster, clusterName, pool, req.Count > 0, newCount, req.Taints != nil || req.Labels != nil, newTaints, newLabels)
+	h.tasks.Go(func() {
+		h.asyncPatchPool(cluster, clusterName, pool, req.Count > 0, newCount, req.Taints != nil || req.Labels != nil, newTaints, newLabels)
+	})
 
 	writeJSON(w, http.StatusAccepted, poolToResponse(pool))
 }
@@ -1071,7 +1079,7 @@ func (h *ClusterHandler) RemoveNodePool(w http.ResponseWriter, r *http.Request)
 		h.log.Error().Err(err).Str("pool", poolName).Msg("mark pool deleting")
 	}
 
-	go h.asyncRemovePool(cluster, clusterID, clusterName, pool)
+	h.tasks.Go(func() { h.asyncRemovePool(cluster, clusterID, clusterName, pool) })
 
 	w.WriteHeader(http.StatusAccepted)
 }

dc-api/internal/api/handlers/dns_zone.go

@@ -23,6 +23,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/wso2/dc-api/internal/api/middleware"
+	"github.com/wso2/dc-api/internal/async"
 	"github.com/wso2/dc-api/internal/db"
 	"github.com/wso2/dc-api/internal/models"
 	"github.com/wso2/dc-api/internal/providers"
@@ -37,12 +38,15 @@ type PrivateDnsZoneHandler struct {
 	resolve       providers.Resolver
 	defaultRegion string
 	defaultZone   string
-	log           zerolog.Logger
+	// tasks tracks the async provisioning goroutines so shutdown can drain
+	// them (bounded). May be nil (tests) — async.Group is nil-receiver-safe.
+	tasks *async.Group
+	log   zerolog.Logger
 }
 
 // NewPrivateDnsZoneHandler creates a PrivateDnsZoneHandler with injected dependencies.
-func NewPrivateDnsZoneHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, log zerolog.Logger) *PrivateDnsZoneHandler {
-	return &PrivateDnsZoneHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, log: log}
+func NewPrivateDnsZoneHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, tasks *async.Group, log zerolog.Logger) *PrivateDnsZoneHandler {
+	return &PrivateDnsZoneHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, tasks: tasks, log: log}
 }
 
 // network resolves the NetworkProvider for the parent VNet's (region, zone).
@@ -253,9 +257,11 @@ func (h *PrivateDnsZoneHandler) CreateZone(w http.ResponseWriter, r *http.Reques
 	}
 
 
-	go h.asyncProvisionZone(network, zone.ID, tenantID, userID, vnet.BackendUID, models.DnsZoneSpec{
-		ZoneName:    req.Name,
-		Description: req.Description,
+	h.tasks.Go(func() {
+		h.asyncProvisionZone(network, zone.ID, tenantID, userID, vnet.BackendUID, models.DnsZoneSpec{
+			ZoneName:    req.Name,
+			Description: req.Description,
+		})
 	})
 
 	w.Header().Set("Content-Type", "application/json")
@@ -373,7 +379,7 @@ func (h *PrivateDnsZoneHandler) DeleteZone(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	go func() {
+	h.tasks.Go(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 		defer cancel()
 		if zone.BackendUID == "" {
@@ -386,7 +392,7 @@ func (h *PrivateDnsZoneHandler) DeleteZone(w http.ResponseWriter, r *http.Reques
 			return
 		}
 		_ = h.repo.DeleteDNSZone(ctx, zoneID)
-	}()
+	})
 
 	w.WriteHeader(http.StatusAccepted)
 }

dc-api/internal/api/handlers/peering.go

@@ -27,6 +27,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/wso2/dc-api/internal/api/middleware"
+	"github.com/wso2/dc-api/internal/async"
 	"github.com/wso2/dc-api/internal/db"
 	"github.com/wso2/dc-api/internal/models"
 	"github.com/wso2/dc-api/internal/placement"
@@ -44,12 +45,15 @@ type PeeringHandler struct {
 	resolve       providers.Resolver
 	defaultRegion string
 	defaultZone   string
-	log           zerolog.Logger
+	// tasks tracks the async provisioning goroutines so shutdown can drain
+	// them (bounded). May be nil (tests) — async.Group is nil-receiver-safe.
+	tasks *async.Group
+	log   zerolog.Logger
 }
 
 // NewPeeringHandler creates a PeeringHandler with injected dependencies.
-func NewPeeringHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, log zerolog.Logger) *PeeringHandler {
-	return &PeeringHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, log: log}
+func NewPeeringHandler(repo *db.Repository, resolve providers.Resolver, defaultRegion, defaultZone string, tasks *async.Group, log zerolog.Logger) *PeeringHandler {
+	return &PeeringHandler{repo: repo, resolve: resolve, defaultRegion: defaultRegion, defaultZone: defaultZone, tasks: tasks, log: log}
 }
 
 // network resolves the NetworkProvider for (region, zone). Empty values resolve
@@ -274,7 +278,9 @@ func (h *PeeringHandler) Create(w http.ResponseWriter, r *http.Request) {
 		PeerAddressSpace:      peerVNet.AddressSpace,
 		TransitCIDR:           transitCIDR,
 	}
-	go h.asyncProvisionPeering(network, peering.ID, tenantID, userID, vnet.BackendUID, peerVNet.BackendUID, spec)
+	h.tasks.Go(func() {
+		h.asyncProvisionPeering(network, peering.ID, tenantID, userID, vnet.BackendUID, peerVNet.BackendUID, spec)
+	})
 
 	resp := peeringToResponse(peering)
 	w.Header().Set("Content-Type", "application/json")
@@ -436,7 +442,7 @@ func (h *PeeringHandler) Delete(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	go func() {
+	h.tasks.Go(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 		defer cancel()
 		if peering.BackendUID == "" {
@@ -459,7 +465,7 @@ func (h *PeeringHandler) Delete(w http.ResponseWriter, r *http.Request) {
 				Msg("release transit CIDR (non-fatal — CASCADE will clean up)")
 		}
 		_ = h.repo.DeletePeering(ctx, peeringID)
-	}()
+	})
 
 	w.WriteHeader(http.StatusAccepted)
 }