Verify admission and bootstrap protocols; fix delayed slot claims

.github/workflows/protocol.yml

@@ -7,6 +7,7 @@ on:
       - 'components/mesh-conn/**'
       - 'components/admission-broker/**'
       - 'components/admission-client/**'
+      - 'components/bootstrap-secrets/**'
       - '.github/workflows/protocol.yml'
   push:
     branches: [main]
@@ -15,13 +16,50 @@ on:
       - 'components/mesh-conn/**'
       - 'components/admission-broker/**'
       - 'components/admission-client/**'
+      - 'components/bootstrap-secrets/**'
       - '.github/workflows/protocol.yml'
   workflow_dispatch:
 
 permissions:
   contents: read
 
 jobs:
+  bootstrap:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    defaults:
+      run:
+        working-directory: components/bootstrap-secrets
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: components/bootstrap-secrets/go.mod
+          cache-dependency-path: components/bootstrap-secrets/go.sum
+      - name: Check allocator regressions and concurrent claims
+        run: go test -race ./...
+      - name: Vet bootstrap secrets
+        run: go vet ./...
+
+  security:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+      - name: Check symbolic security properties and expected attacks
+        run: python3 protocol/security/check.py
+      - name: Save symbolic security evidence
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: protocol-security-checks
+          path: .local/protocol-security/
+          include-hidden-files: true
+          if-no-files-found: warn
+
   tlc:
     runs-on: ubuntu-latest
     timeout-minutes: 10
@@ -53,7 +91,7 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: '3.12'
-      - name: Check unbounded signaling proof and all axiom dependencies
+      - name: Check protocol proofs and all axiom dependencies
         run: python3 protocol/proof/check.py
       - name: Save Lean evidence
         if: always()

CHANGELOG.md

@@ -6,6 +6,10 @@ This changelog tracks `service-mesh` releases.
 
 ### Changed
 
+- Model Cloud API read retries, quota waits, cancellation, and deployment
+  deadlines, including controls that reproduce failure without client retries.
+  Link the focused Go SDK fix that reuses existing retries for Terraform reads.
+
 - Add `modules/mesh-coordinators` so examples keep env values beside named
   instances while the module handles broker discovery and separate value
   resources. Include state moves from the existing flat coordinator resources.
@@ -47,6 +51,10 @@ This changelog tracks `service-mesh` releases.
 
 ### Security
 
+- Fixed dynamic slot allocation to reread a slot after failed CAS, preventing
+  delayed requests across restarts from assigning one instance two slots.
+  Use consistent reads and report exhausted permanent slots without waiting
+  for unsafe deletion; slot reuse remains unsupported.
 - Patroni passwords now come from in-TEE derivation, not Terraform
   state.
 - Validated the experimental KMS key-provider setting on AMD SEV-SNP so
@@ -60,6 +68,12 @@ This changelog tracks `service-mesh` releases.
 
 ### Documentation
 
+- Added kernel-checked admission-policy selection and scope proofs, including
+  wildcard precedence and optional measurement pins; extended admission models
+  to brokers sharing a platform authority and added slot-lifecycle traces.
+- Added unbounded symbolic admission and signaling-authentication checks,
+  expected attack traces, and a protocol security roadmap that makes
+  shared-key, instance-identity, and token-channel assumptions explicit.
 - Added kernel-checked Lean proofs of unbounded signaling-state convergence
   and candidate dissemination, with explicit failure/timing premises and
   executable NAT-expiry counterexamples.

components/bootstrap-secrets/main.go

@@ -284,10 +284,10 @@ func mustEnv(k string) string {
 //   - a slot whose value is already this InstanceID (we're rejoining), or
 //   - the lowest empty slot (CAS-claim it).
 //
-// First match wins. Returns the slot index. Slot ownership is
-// permanent for the InstanceID's lifetime; cleanup of stale slots
-// (when an instance is permanently retired) is a separate operator
-// task — see the example README.
+// First match wins. Returns the slot index. Ownership records are permanent:
+// deletion or reassignment is unsupported, even after an instance retires,
+// because its outstanding CAS requests may still commit. Safe reuse requires
+// a separate protocol that fences both old owners and outstanding requests.
 func claimOrdinal(cfg *Config, instanceID string) (int, error) {
 	if cfg.ConsulHTTPAddr == "" {
 		return 0, fmt.Errorf("CONSUL_HTTP_ADDR required for non-coordinator role")
@@ -303,46 +303,39 @@ func claimOrdinal(cfg *Config, instanceID string) (int, error) {
 
 	keyPrefix := fmt.Sprintf("cluster/%s/slots", cfg.ClusterName)
 
-	// Retry the whole walk a few times — pollLoop racing with peers
-	// could cause CAS misses; on a miss, try the next slot or restart
-	// the walk.
-	for attempt := 0; attempt < 20; attempt++ {
-		for i := 0; i < cfg.ExpectedReplicas; i++ {
-			key := fmt.Sprintf("%s/%d", keyPrefix, i)
-			existing, _, err := kv.Get(key, nil)
+	for i := 0; i < cfg.ExpectedReplicas; i++ {
+		key := fmt.Sprintf("%s/%d", keyPrefix, i)
+		for {
+			existing, _, err := kv.Get(key, &consulapi.QueryOptions{RequireConsistent: true})
 			if err != nil {
 				return 0, fmt.Errorf("kv get %s: %w", key, err)
 			}
-			switch {
-			case existing != nil && string(existing.Value) == instanceID:
-				log.Printf("rejoining slot %d (already owned)", i)
-				return i, nil
-			case existing == nil:
-				ok, _, err := kv.CAS(&consulapi.KVPair{
-					Key:         key,
-					Value:       []byte(instanceID),
-					ModifyIndex: 0,
-				}, nil)
-				if err != nil {
-					return 0, fmt.Errorf("kv cas %s: %w", key, err)
-				}
-				if ok {
-					log.Printf("claimed slot %d (fresh)", i)
+			if existing != nil {
+				if string(existing.Value) == instanceID {
+					log.Printf("rejoining slot %d (already owned)", i)
 					return i, nil
 				}
-				// CAS lost the race; some other peer claimed
-				// this slot first. Try the next slot.
-			default:
-				// owned by another instance; skip
+				// Only observing another owner permits advancing the scan.
+				break
+			}
+			ok, _, err := kv.CAS(&consulapi.KVPair{
+				Key:         key,
+				Value:       []byte(instanceID),
+				ModifyIndex: 0,
+			}, nil)
+			if err != nil {
+				return 0, fmt.Errorf("kv cas %s: %w", key, err)
+			}
+			if ok {
+				log.Printf("claimed slot %d (fresh)", i)
+				return i, nil
 			}
+			// A delayed request from our own previous incarnation may
+			// have claimed this slot. Reread it before deciding whether
+			// to rejoin or advance; CAS failure alone cannot tell us.
 		}
-		// Exhausted slots without claiming or rejoining; either the
-		// cluster is over-replicated or there's a stale slot. Sleep
-		// briefly and retry — gives a slot a chance to clear if a
-		// peer is in transient state.
-		time.Sleep(2 * time.Second)
 	}
-	return 0, fmt.Errorf("no available slot in cluster %q (max=%d) — cluster over-replicated or has stale slots",
+	return 0, fmt.Errorf("no available slot in cluster %q (max=%d); dynamic slot ownership is permanent",
 		cfg.ClusterName, cfg.ExpectedReplicas)
 }
 

components/bootstrap-secrets/main_test.go

@@ -0,0 +1,206 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"maps"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+
+	consulapi "github.com/hashicorp/consul/api"
+)
+
+// slotRegistry implements the Consul operations used by claimOrdinal. The
+// delayedOwner hook commits an earlier caller's request just before the first
+// CAS is evaluated, without relying on process timing or sleeps.
+type slotRegistry struct {
+	mu           sync.Mutex
+	owners       map[string]string
+	delayedOwner string
+	requests     []string
+	failMethod   string
+}
+
+func (r *slotRegistry) serveHTTP(w http.ResponseWriter, req *http.Request) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	key := strings.TrimPrefix(req.URL.Path, "/v1/kv/")
+	r.requests = append(r.requests, req.Method+" "+key)
+	if req.Method == r.failMethod {
+		http.Error(w, "registry unavailable", http.StatusServiceUnavailable)
+		return
+	}
+	switch req.Method {
+	case http.MethodGet:
+		if !req.URL.Query().Has("consistent") {
+			http.Error(w, "expected strongly consistent read", http.StatusBadRequest)
+			return
+		}
+		owner, exists := r.owners[key]
+		if !exists {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		_ = json.NewEncoder(w).Encode([]*consulapi.KVPair{{Key: key, Value: []byte(owner), ModifyIndex: 1}})
+	case http.MethodPut:
+		if req.URL.Query().Get("cas") != "0" {
+			http.Error(w, "expected create-if-absent CAS", http.StatusBadRequest)
+			return
+		}
+		body, err := io.ReadAll(req.Body)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		if r.delayedOwner != "" {
+			r.owners[key] = r.delayedOwner
+			r.delayedOwner = ""
+		}
+		_, exists := r.owners[key]
+		if !exists {
+			r.owners[key] = string(body)
+		}
+		_ = json.NewEncoder(w).Encode(!exists)
+	default:
+		http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+	}
+}
+
+func (r *slotRegistry) snapshot() (map[string]string, []string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return maps.Clone(r.owners), slices.Clone(r.requests)
+}
+
+func TestClaimOrdinalDelayedCAS(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		delayedOwner string
+		wantOrdinal  int
+		wantRequests []string
+	}{
+		{
+			name:         "old incarnation claims for this instance",
+			delayedOwner: "this-instance",
+			wantOrdinal:  0,
+			wantRequests: []string{"GET cluster/test/slots/0", "PUT cluster/test/slots/0", "GET cluster/test/slots/0"},
+		},
+		{
+			name:         "another instance claims the slot",
+			delayedOwner: "other-instance",
+			wantOrdinal:  1,
+			wantRequests: []string{"GET cluster/test/slots/0", "PUT cluster/test/slots/0", "GET cluster/test/slots/0", "GET cluster/test/slots/1", "PUT cluster/test/slots/1"},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			registry := &slotRegistry{owners: make(map[string]string), delayedOwner: tc.delayedOwner}
+			server := httptest.NewServer(http.HandlerFunc(registry.serveHTTP))
+			defer server.Close()
+			ordinal, err := claimOrdinal(&Config{ClusterName: "test", ConsulHTTPAddr: server.URL, ExpectedReplicas: 2}, "this-instance")
+			if err != nil {
+				t.Fatal(err)
+			}
+			owners, requests := registry.snapshot()
+			if ordinal != tc.wantOrdinal {
+				t.Fatalf("ordinal = %d, want %d; registry = %v", ordinal, tc.wantOrdinal, owners)
+			}
+			if !reflect.DeepEqual(requests, tc.wantRequests) {
+				t.Fatalf("requests = %v, want %v", requests, tc.wantRequests)
+			}
+		})
+	}
+}
+
+func TestClaimOrdinalRejoinsPermanentSlot(t *testing.T) {
+	registry := &slotRegistry{owners: map[string]string{"cluster/test/slots/0": "other-instance", "cluster/test/slots/1": "this-instance"}}
+	server := httptest.NewServer(http.HandlerFunc(registry.serveHTTP))
+	defer server.Close()
+	ordinal, err := claimOrdinal(&Config{ClusterName: "test", ConsulHTTPAddr: server.URL, ExpectedReplicas: 2}, "this-instance")
+	if err != nil || ordinal != 1 {
+		t.Fatalf("claimOrdinal = (%d, %v), want (1, nil)", ordinal, err)
+	}
+	_, requests := registry.snapshot()
+	if len(requests) != 2 {
+		t.Fatalf("rejoining must only read the registry, got %v", requests)
+	}
+}
+
+func TestClaimOrdinalFullRegistry(t *testing.T) {
+	registry := &slotRegistry{owners: map[string]string{"cluster/test/slots/0": "other-instance", "cluster/test/slots/1": "another-instance"}}
+	server := httptest.NewServer(http.HandlerFunc(registry.serveHTTP))
+	defer server.Close()
+	_, err := claimOrdinal(&Config{ClusterName: "test", ConsulHTTPAddr: server.URL, ExpectedReplicas: 2}, "this-instance")
+	if err == nil || !strings.Contains(err.Error(), "no available slot") {
+		t.Fatalf("expected exhausted registry error, got %v", err)
+	}
+	_, requests := registry.snapshot()
+	if len(requests) != 2 {
+		t.Fatalf("permanent ownership cannot clear on a retry, got %v", requests)
+	}
+}
+
+func TestClaimOrdinalRegistryErrors(t *testing.T) {
+	for _, method := range []string{http.MethodGet, http.MethodPut} {
+		t.Run(method, func(t *testing.T) {
+			registry := &slotRegistry{owners: make(map[string]string), failMethod: method}
+			server := httptest.NewServer(http.HandlerFunc(registry.serveHTTP))
+			defer server.Close()
+			_, err := claimOrdinal(&Config{ClusterName: "test", ConsulHTTPAddr: server.URL, ExpectedReplicas: 2}, "this-instance")
+			if err == nil || !strings.Contains(err.Error(), "registry unavailable") {
+				t.Fatalf("expected registry error, got %v", err)
+			}
+			_, requests := registry.snapshot()
+			if len(requests) > 2 {
+				t.Fatalf("storage errors must terminate the claim, got %v", requests)
+			}
+		})
+	}
+}
+
+func TestClaimOrdinalConcurrentInstances(t *testing.T) {
+	const replicas = 8
+	registry := &slotRegistry{owners: make(map[string]string)}
+	server := httptest.NewServer(http.HandlerFunc(registry.serveHTTP))
+	defer server.Close()
+	cfg := &Config{ClusterName: "test", ConsulHTTPAddr: server.URL, ExpectedReplicas: replicas}
+	type result struct {
+		instance string
+		ordinal  int
+		err      error
+	}
+	start := make(chan struct{})
+	results := make(chan result, replicas)
+	for i := 0; i < replicas; i++ {
+		go func() {
+			<-start
+			instance := fmt.Sprintf("instance-%d", i)
+			ordinal, err := claimOrdinal(cfg, instance)
+			results <- result{instance, ordinal, err}
+		}()
+	}
+	close(start)
+	claims := make(map[int]string)
+	for i := 0; i < replicas; i++ {
+		claim := <-results
+		if claim.err != nil {
+			t.Error(claim.err)
+			continue
+		}
+		if previous, exists := claims[claim.ordinal]; exists {
+			t.Errorf("instances %s and %s both claimed %d", previous, claim.instance, claim.ordinal)
+		}
+		claims[claim.ordinal] = claim.instance
+	}
+	owners, _ := registry.snapshot()
+	for ordinal, instance := range claims {
+		if got := owners[fmt.Sprintf("cluster/test/slots/%d", ordinal)]; got != instance {
+			t.Errorf("slot %d owner = %q, want %q", ordinal, got, instance)
+		}
+	}
+}