Address self-review findings on the plumbing slice

Pre-CodeRabbit review pass (the local pr-review gate) on this branch:

- Declare StatusFallbackOK: false explicitly on the vpc and subnet
  capabilities — their Get consumers are full-object read-modify-
  writes, so the status-only degrade must never engage; spell out the
  zero value the way vmCapability spells out its true.
- Add VerbIsRoutable (direct lookup, no map copy) and use it in the
  Get/List/Apply routability guards.
- Explain why Create/Update are exempt from Apply's guard, and add
  the missing in-method comment on DisassociateRouteTable.
- Test gaps: force=false forwarding, Apply wire-error translation,
  Apply unmapped-GVR refusal; seam Get/Apply failure propagation for
  the plumbing ops (no apply after a failed read); an explicit
  assertion that the legacy CIDR-fallback List engaged in the
  fail-closed test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RC3BDHCFmJBdxvGVW69FAr

Author: HiranAdikari

dc-api/internal/providers/clusteraccess/agent.go

@@ -134,7 +134,7 @@ func (a *AgentBacked) Get(ctx context.Context, gvr schema.GroupVersionResource,
 	// routable family for Get — e.g. the pre-seeded virtualmachineinstances, which
 	// has no RouteVerbs. Refuse rather than issue a read the agent's SA may not be
 	// permitted to serve; the caller falls back to the direct path.
-	if verbs, ok := RoutableVerbs(gvr); !ok || !verbs[VerbGet] {
+	if !VerbIsRoutable(gvr, VerbGet) {
 		return nil, fmt.Errorf("clusteraccess: get of %s not routable: %w", gvr.String(), agentgw.ErrOpNotRoutable)
 	}
 	ref, err := a.ref(gvr, ns, name)
@@ -250,7 +250,7 @@ func (a *AgentBacked) List(ctx context.Context, gvr schema.GroupVersionResource,
 	// virtualmachineinstances, which is in the mapper as a superset but has no
 	// RouteVerbs. Refuse to route such a family to the agent rather than issuing a
 	// list the agent's SA may not be permitted (least-privilege) to serve.
-	if verbs, ok := RoutableVerbs(gvr); !ok || !verbs[VerbList] {
+	if !VerbIsRoutable(gvr, VerbList) {
 		return nil, fmt.Errorf("clusteraccess: list of %s not routable: %w", gvr.String(), agentgw.ErrOpNotRoutable)
 	}
 	apiVersion, kind, ok := a.mapper.GVK(gvr)
@@ -333,9 +333,13 @@ func (a *AgentBacked) Apply(ctx context.Context, gvr schema.GroupVersionResource
 	// for a family that does not declare VerbApply, rather than performing a write
 	// the allow-set never granted. The Routed decision already gates this per
 	// call; the guard keeps the contract uniform for any direct AgentBacked use.
-	// Create/Update do NOT pass through here (they call applyWith directly), so
-	// families that route VerbCreate without VerbApply are unaffected.
-	if verbs, ok := RoutableVerbs(gvr); !ok || !verbs[VerbApply] {
+	// Create/Update are DELIBERATELY exempt: they call applyWith directly, are
+	// gated per call by the Routed decision's verb allow-set (VerbCreate/
+	// VerbUpdate), always use the accessor's own fieldManager, and have no direct
+	// AgentBacked callers — so a family that routes VerbCreate without VerbApply
+	// (SA/RoleBinding, provider-network-style creates) still creates via SSA
+	// without needing VerbApply in its routable set.
+	if !VerbIsRoutable(gvr, VerbApply) {
 		return nil, fmt.Errorf("clusteraccess: apply of %s not routable: %w", gvr.String(), agentgw.ErrOpNotRoutable)
 	}
 	if fieldManager == "" {

dc-api/internal/providers/clusteraccess/capabilities.go

@@ -156,6 +156,11 @@ var (
 		Namespaced: false,
 		RouteVerbs: []Verb{VerbGet, VerbCreate, VerbApply, VerbDelete},
 		AgentVerbs: []Verb{VerbGet, VerbList, VerbWatch, VerbCreate, VerbApply, VerbDelete},
+		// Explicitly false (the zero value, spelled out like vmCapability's explicit
+		// true): the Vpc Get consumers are full-object read-modify-writes — the
+		// peering (spec.vpcPeerings) and static-route (spec.staticRoutes) ops read
+		// .spec to rebuild it — so the status-only degrade must never engage.
+		StatusFallbackOK: false,
 	}
 	// subnetCapability onboards the Subnet CRUD that CreateSubnet/GetSubnet/
 	// DeleteSubnet route through the kubeovn seam, plus the spec-write plumbing
@@ -169,6 +174,10 @@ var (
 		Namespaced: false,
 		RouteVerbs: []Verb{VerbGet, VerbCreate, VerbApply, VerbDelete},
 		AgentVerbs: []Verb{VerbGet, VerbList, VerbWatch, VerbCreate, VerbApply, VerbDelete},
+		// Explicitly false (mirrors vpcCapability): the Subnet Get consumers are
+		// full-object read-modify-writes — the NSG ACL (spec.acls) ops read .spec
+		// to rebuild it — so the status-only degrade must never engage.
+		StatusFallbackOK: false,
 	}
 	// serviceAccountCapability onboards the ServiceAccount create that
 	// EnsureCloudProviderSA routes through the harvester seam (the cloud-provider
@@ -323,6 +332,15 @@ func RoutableVerbs(gvr schema.GroupVersionResource) (map[Verb]bool, bool) {
 	return out, true
 }
 
+// VerbIsRoutable reports whether a single verb is in a GVR's routable set — a
+// direct lookup with no defensive map copy, for the per-call guards on the hot
+// read/write paths (AgentBacked.Get/List/Apply). Callers that need the whole
+// set keep using RoutableVerbs.
+func VerbIsRoutable(gvr schema.GroupVersionResource, v Verb) bool {
+	buildDerived()
+	return derivedRouteSet[gvr][v]
+}
+
 // StatusFallbackOK reports whether AgentBacked.Get may degrade a routed read of
 // this GVR to a status-only get_status when an older agent lacks get_object. True
 // only for families whose Get consumers read status alone (VMs); false (the

dc-api/internal/providers/clusteraccess/clusteraccess_test.go

@@ -581,13 +581,62 @@ func TestAgentBackedApply_ForwardsCallerFieldManager(t *testing.T) {
 		t.Error("wire force = false, want the caller's true")
 	}
 
-	// Empty caller manager → the accessor default.
+	// Empty caller manager → the accessor default; force=false forwards too.
 	if _, err := a.Apply(context.Background(), vpcGVR, "", obj, "", false); err != nil {
 		t.Fatalf("Apply (empty manager) error: %v", err)
 	}
 	if gotFM != "dc-api" {
 		t.Errorf("wire fieldManager = %q for an empty caller value, want the accessor default %q", gotFM, "dc-api")
 	}
+	if gotForce {
+		t.Error("wire force = true for the second call, want the caller's false")
+	}
+}
+
+// TestAgentBackedApply_SessionErrorTranslated mirrors the Get/List
+// agent-unavailable tests: a wire failure from Session.Apply must surface
+// through translateErr so callers' errors.Is(ErrAgentUnavailable) checks fire.
+func TestAgentBackedApply_SessionErrorTranslated(t *testing.T) {
+	vpcGVR := schema.GroupVersionResource{Group: "kubeovn.io", Version: "v1", Resource: "vpcs"}
+	sess := &fakeSession{apply: func(json.RawMessage, string, bool) (agentgw.ApplyResult, error) {
+		return agentgw.ApplyResult{}, agentgw.ErrAgentUnavailable
+	}}
+	a := NewAgentBacked(sess, "lk", "zone-1", "dc-api", DefaultGVKMapper(), zerolog.Nop())
+
+	obj := &unstructured.Unstructured{Object: map[string]interface{}{
+		"apiVersion": "kubeovn.io/v1", "kind": "Vpc",
+		"metadata": map[string]interface{}{"name": "vnet-a"},
+	}}
+	_, err := a.Apply(context.Background(), vpcGVR, "", obj, "dc-api-kubeovn-staticroutes", true)
+	if !errors.Is(err, agentgw.ErrAgentUnavailable) {
+		t.Errorf("error does not wrap ErrAgentUnavailable: %v", err)
+	}
+}
+
+// TestAgentBackedApply_UnmappedGVRNotRoutable mirrors the Get/List unmapped-GVR
+// tests: an apply for a GVR outside the capability registry is refused with
+// ErrOpNotRoutable before any agent op is issued (the routability guard fires
+// first — an unmapped GVR is by definition not Apply-routable).
+func TestAgentBackedApply_UnmappedGVRNotRoutable(t *testing.T) {
+	bogusGVR := schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "widgets"}
+	applyCalled := false
+	sess := &fakeSession{apply: func(json.RawMessage, string, bool) (agentgw.ApplyResult, error) {
+		applyCalled = true
+		return agentgw.ApplyResult{}, nil
+	}}
+	a := NewAgentBacked(sess, "lk", "zone-1", "dc-api", DefaultGVKMapper(), zerolog.Nop())
+
+	obj := &unstructured.Unstructured{Object: map[string]interface{}{
+		"apiVersion": "example.com/v1", "kind": "Widget",
+		"metadata": map[string]interface{}{"name": "w"},
+	}}
+	_, err := a.Apply(context.Background(), bogusGVR, "", obj, "dc-api", true)
+	if !errors.Is(err, agentgw.ErrOpNotRoutable) {
+		t.Errorf("error = %v, want ErrOpNotRoutable", err)
+	}
+	if applyCalled {
+		t.Error("agent Apply was issued for an unmapped GVR")
+	}
 }
 
 // TestAgentBackedApply_NotRoutableFamilyRefused mirrors the Get/List guards: an

dc-api/internal/providers/kubeovn/client.go

@@ -1021,6 +1021,9 @@ func (c *Client) AssociateRouteTable(_ context.Context, _, _ string) error {
 
 // DisassociateRouteTable is a no-op for the same reason as AssociateRouteTable.
 func (c *Client) DisassociateRouteTable(_ context.Context, _, _ string) error {
+	// M2 stance (a): routes apply VPC-wide.  No backend change (and therefore no
+	// c.dynamic guard — a remote zone's disassociation is the same pure no-op).
+	// See AssociateRouteTable.
 	return nil
 }
 

dc-api/internal/providers/kubeovn/network_plumbing_seam_test.go

@@ -30,6 +30,7 @@ package kubeovn
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
 	"sync"
 	"testing"
@@ -69,6 +70,11 @@ type plumbRecordingAccessor struct {
 
 	listCalls int
 	listErr   error
+
+	// getErr/applyErr, when set, make every Get/Apply fail — for asserting the
+	// driver propagates seam errors and stops (no apply after a failed read).
+	getErr   error
+	applyErr error
 }
 
 var _ clusteraccess.Accessor = (*plumbRecordingAccessor)(nil)
@@ -77,6 +83,9 @@ func (a *plumbRecordingAccessor) Get(_ context.Context, gvr schema.GroupVersionR
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	a.getCalls = append(a.getCalls, gvr)
+	if a.getErr != nil {
+		return nil, a.getErr
+	}
 	obj, ok := a.objects[name]
 	if !ok {
 		return nil, k8serrors.NewNotFound(gvr.GroupResource(), name)
@@ -101,6 +110,9 @@ func (a *plumbRecordingAccessor) Create(_ context.Context, _ schema.GroupVersion
 func (a *plumbRecordingAccessor) Apply(_ context.Context, gvr schema.GroupVersionResource, ns string, obj *unstructured.Unstructured, fieldManager string, force bool) (*unstructured.Unstructured, error) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
+	if a.applyErr != nil {
+		return nil, a.applyErr
+	}
 	// Store the object as-is (no DeepCopy): the driver builds list entries from
 	// model structs whose ints are plain `int`, which the unstructured JSON
 	// deep-copier rejects — json.Marshal (what both real seams do) handles them
@@ -690,12 +702,16 @@ func TestRemoteClient_LegacyPeeringCIDRFallback_FailsClosed(t *testing.T) {
 	// Mirror buildRemoteSet: Routed whose Direct fallback is NoCreds, decision
 	// routing only the verbs the plumbing families declare (Get/Apply here; List
 	// deliberately absent — mirroring subnetCapability.RouteVerbs).
+	sawListRequest := false
 	routed := clusteraccess.NewRouted(
 		clusteraccess.NewNoCreds("lk", "zone-2"),
 		func(v clusteraccess.Verb, _ schema.GroupVersionResource) (clusteraccess.Accessor, bool) {
 			switch v {
 			case clusteraccess.VerbGet, clusteraccess.VerbApply:
 				return agent, true
+			case clusteraccess.VerbList:
+				sawListRequest = true // the legacy CIDR fallback engaged
+				return nil, false
 			default:
 				return nil, false
 			}
@@ -711,4 +727,75 @@ func TestRemoteClient_LegacyPeeringCIDRFallback_FailsClosed(t *testing.T) {
 	if !strings.Contains(err.Error(), "no agent connected for zone lk/zone-2") {
 		t.Errorf("error = %q, want the NoCreds fail-closed message naming the zone", err.Error())
 	}
+	// Explicit: the failure came from the CIDR-fallback List actually engaging —
+	// a refactor that silently skips the fallback must not pass this test.
+	if !sawListRequest {
+		t.Error("the legacy subnet-CIDR fallback List was never requested — the fail-closed path did not engage")
+	}
+}
+
+// ── Seam error propagation (Get/Apply failures) ──────────────────────────────
+
+// TestAppendVpcPeering_GetErrorPropagates_NoApply: a seam Get failure aborts the
+// read-modify-apply before any write — the error propagates and Apply is never
+// issued (no blind write on unknown state).
+func TestAppendVpcPeering_GetErrorPropagates_NoApply(t *testing.T) {
+	boom := fmt.Errorf("seam get exploded")
+	acc := &plumbRecordingAccessor{getErr: boom}
+	c := &Client{access: acc}
+
+	err := c.appendVpcPeering(context.Background(), "vnet-a", "vnet-b", "100.64.10.0/24")
+	if err == nil || !strings.Contains(err.Error(), "seam get exploded") {
+		t.Fatalf("appendVpcPeering must propagate the Get error, got %v", err)
+	}
+	if got := acc.applyCount(); got != 0 {
+		t.Errorf("seam Apply called %d times after a failed Get, want 0", got)
+	}
+}
+
+// TestAppendVpcPeering_ApplyErrorPropagates: a seam Apply failure propagates to
+// the caller (the handler surfaces it; nothing swallows the write error).
+func TestAppendVpcPeering_ApplyErrorPropagates(t *testing.T) {
+	boom := fmt.Errorf("seam apply exploded")
+	acc := &plumbRecordingAccessor{
+		objects: map[string]*unstructured.Unstructured{
+			"vnet-a": seededVpc("vnet-a", []interface{}{}, []interface{}{}),
+		},
+		applyErr: boom,
+	}
+	c := &Client{access: acc}
+
+	err := c.appendVpcPeering(context.Background(), "vnet-a", "vnet-b", "100.64.10.0/24")
+	if err == nil || !strings.Contains(err.Error(), "seam apply exploded") {
+		t.Fatalf("appendVpcPeering must propagate the Apply error, got %v", err)
+	}
+}
+
+// TestPatchSubnetACLs_ApplyErrorPropagates: the pure-apply subnet-ACL op (its
+// callers do the feeding Get) propagates a seam Apply failure unchanged.
+func TestPatchSubnetACLs_ApplyErrorPropagates(t *testing.T) {
+	boom := fmt.Errorf("seam apply exploded")
+	acc := &plumbRecordingAccessor{applyErr: boom}
+	c := &Client{access: acc}
+	err := c.patchSubnetACLs(context.Background(), "subnet-1", []interface{}{})
+	if err == nil || !strings.Contains(err.Error(), "seam apply exploded") {
+		t.Fatalf("patchSubnetACLs must propagate the Apply error, got %v", err)
+	}
+}
+
+// TestUpdateRouteTableRoutes_GetErrorPropagates_NoApply: the route-table op's
+// FEEDING read fails → the error propagates and the write never runs (same
+// no-blind-write property as the peering op, on the routes code path).
+func TestUpdateRouteTableRoutes_GetErrorPropagates_NoApply(t *testing.T) {
+	boom := fmt.Errorf("seam get exploded")
+	acc := &plumbRecordingAccessor{getErr: boom}
+	c := &Client{access: acc}
+
+	err := c.UpdateRouteTableRoutes(context.Background(), "vnet-a/rt-11111111-1111-1111-1111-111111111111", nil)
+	if err == nil || !strings.Contains(err.Error(), "seam get exploded") {
+		t.Fatalf("UpdateRouteTableRoutes must propagate the Get error, got %v", err)
+	}
+	if got := acc.applyCount(); got != 0 {
+		t.Errorf("seam Apply called %d times after a failed Get, want 0", got)
+	}
 }