fix(masque): Keep a connector's /96 when its grant record fails

docs/edge/masque-authority.md

@@ -73,6 +73,8 @@ masque-authority grant retire  <connector>   # revoke and tombstone the allocati
 
 `ledger init` refuses ranges that aren't masked IPv6 /80s inside one of the VPC's networks, or that overlap each other. It also refuses ranges that overlap those of another cell's ledger for the same VPC, so two cells are never handed overlapping /80s. And it refuses while any EdgeService the authority issued for the VPC in that cell still exists, so a recreated ledger can't hand out a /96 or pool that one still holds. Issued services of other cells don't block it. Choose ranges that overlap no workload subnet; core refuses a delegated route that does.
 
+`grant create` allocates the /96 before it writes the grant record. If the record write fails, the error names the connector and its /96. Run `grant create` again with `--connector <connector>` to keep that /96, or `grant retire <connector>` to tombstone it. A plain retry allocates another /96. If the error says the connector already has a grant record, the write succeeded but its reply was lost: run `grant rotate <connector>` for a credential.
+
 Run the server with:
 
 ```sh

docs/edge/masque-provider.md

@@ -26,6 +26,8 @@ Only then does the request get its packet path. The request waits up to 30 secon
 
 Connectors of one service share its attachment and device.
 
+The route is the connector's whole /96, and it is the only route the connector gets. The client numbers addresses and longer subprefixes inside the /96 itself. The device delivers a packet for any address inside the /96 to that connector's request, so no subdivision needs its own route or allocation.
+
 ## Replacing a connector's request
 
 A connector forwards through at most one request. A new request for a connector that already has one open, such as a reconnect or an HTTP/3 to HTTP/2 fallback, replaces it:

internal/edge/masque/authority/grants.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/google/uuid"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -68,6 +69,11 @@ func (g Grants) expiry(lifetime time.Duration) (time.Time, error) {
 // Create allocates the connector's /96 in the ledger of its VPC and cell,
 // writes its grant record with a new credential and returns that
 // credential. It refuses a connector that already has a record.
+//
+// The allocation is written before the record. When the record write
+// fails, the error names the connector: creating it again with the same
+// Connector reuses its allocation rather than taking another /96, and
+// Retire tombstones it.
 func (g Grants) Create(ctx context.Context, req CreateRequest) (Issued, error) {
 	expires, err := g.expiry(req.Lifetime)
 	if err != nil {
@@ -98,7 +104,12 @@ func (g Grants) Create(ctx context.Context, req CreateRequest) (Issued, error) {
 		return Issued{}, fmt.Errorf("allocate connector %s: %w", r.Connector, err)
 	}
 	if err := g.Client.Create(ctx, s); err != nil {
-		return Issued{}, fmt.Errorf("create grant record %s: %w", s.Name, err)
+		if apierrors.IsAlreadyExists(err) {
+			return Issued{}, fmt.Errorf("connector %s already has a grant record; rotate it for a new credential: %w",
+				r.Connector, err)
+		}
+		return Issued{}, fmt.Errorf("create grant record of connector %s, which holds %s; create it again with "+
+			"this connector to keep that /96, or retire it: %w", r.Connector, e.Prefix, err)
 	}
 	return Issued{Connector: r.Connector, Credential: cred, Revision: r.Revision, VPC: r.VPC.Selector(),
 		Prefix: e.Prefix, Expires: expires}, nil
@@ -153,9 +164,13 @@ func (g Grants) Disable(ctx context.Context, connector string) error {
 }
 
 // Retire disables the connector and tombstones its allocation, so its /96
-// and attachment pool are never allocated again.
+// and attachment pool are never allocated again. A connector whose record
+// was never written, because Create failed after allocating, has its
+// allocation tombstoned in whichever of the authority's ledgers holds it.
 func (g Grants) Retire(ctx context.Context, connector string) error {
-	if err := g.Disable(ctx, connector); err != nil {
+	if err := g.Disable(ctx, connector); apierrors.IsNotFound(err) {
+		return g.retireUnrecorded(ctx, connector, err)
+	} else if err != nil {
 		return err
 	}
 	r, _, err := readRecord(ctx, g.Client, g.Namespace, connector)
@@ -165,6 +180,30 @@ func (g Grants) Retire(ctx context.Context, connector string) error {
 	return g.ledgers().Retire(ctx, r.VPC.UID, r.Cell, connector)
 }
 
+// retireUnrecorded tombstones the allocations of connector, which has no
+// record. notFound is the error reading the record returned, for a
+// connector no ledger holds either.
+func (g Grants) retireUnrecorded(ctx context.Context, connector string, notFound error) error {
+	held, err := g.ledgers().holding(ctx, connector)
+	if err != nil {
+		return err
+	}
+	if len(held) == 0 {
+		return notFound
+	}
+	for _, l := range held {
+		if err := g.ledgers().Retire(ctx, l.VPC.UID, l.Cell, connector); err != nil {
+			return err
+		}
+	}
+	// A Create that wrote the record since Disable found none leaves it
+	// enabled; disable it as Retire does a recorded connector.
+	if err := g.Disable(ctx, connector); err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+	return nil
+}
+
 // update applies change to a fresh read of the connector's record and
 // writes it back conditioned on the read's UID and resourceVersion.
 func (g Grants) update(ctx context.Context, connector string, change func(*Record) error) error {

internal/edge/masque/authority/grants_test.go

@@ -0,0 +1,163 @@
+// Copyright 2026 Datum Cloud, Inc.
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package authority
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+)
+
+// recordWrites decides the outcome of each grant record create: nil
+// writes it, errLost writes it and still fails, any other error fails
+// without writing.
+type recordWrites func() error
+
+var errLost = errors.New("reply lost")
+
+// faultyGrants returns Grants over a fake API holding one VPC and its
+// ledger, whose grant record creates go through writes.
+func faultyGrants(t *testing.T, writes recordWrites) (Grants, VPCRef) {
+	t.Helper()
+	vpc := testVPC()
+	c := fake.NewClientBuilder().WithScheme(testScheme()).WithObjects(vpc).WithInterceptorFuncs(interceptor.Funcs{
+		Create: func(ctx context.Context, c client.WithWatch, o client.Object, opts ...client.CreateOption) error {
+			if _, ok := o.(*corev1.Secret); !ok {
+				return c.Create(ctx, o, opts...)
+			}
+			switch err := writes(); {
+			case err == nil:
+				return c.Create(ctx, o, opts...)
+			case errors.Is(err, errLost):
+				if cerr := c.Create(ctx, o, opts...); cerr != nil {
+					return cerr
+				}
+				return err
+			default:
+				return err
+			}
+		},
+	}).Build()
+	g := Grants{Client: c, Namespace: testNS, Now: time.Now}
+	ref := VPCRef{Namespace: tenantNS, Name: vpcName, UID: string(vpc.UID)}
+	if err := g.ledgers().Init(context.Background(), ref, testCell, testConnectors, testPools); err != nil {
+		t.Fatal(err)
+	}
+	return g, VPCRef{Namespace: tenantNS, Name: vpcName}
+}
+
+// entries returns the test VPC's ledger entries.
+func (g Grants) entries(t *testing.T) []Entry {
+	t.Helper()
+	l, _, err := g.ledgers().Read(context.Background(), string(testVPC().UID), testCell)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return l.Entries
+}
+
+// failOnce fails the first record write without writing it.
+func failOnce() recordWrites {
+	n := 0
+	return func() error {
+		if n++; n == 1 {
+			return errors.New("injected")
+		}
+		return nil
+	}
+}
+
+// TestCreateRetryKeepsAllocation: a record write that fails leaves the
+// connector's /96 allocated, and creating the same connector again uses
+// that /96 rather than another.
+func TestCreateRetryKeepsAllocation(t *testing.T) {
+	g, ref := faultyGrants(t, failOnce())
+	ctx := context.Background()
+	_, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Lifetime: time.Hour})
+	if err == nil {
+		t.Fatal("create with a failed record write succeeded")
+	}
+	e := g.entries(t)
+	if len(e) != 1 || !strings.Contains(err.Error(), e[0].Connector) ||
+		!strings.Contains(err.Error(), e[0].Prefix.String()) {
+		t.Fatalf("failed create = %v with entries %+v; want an error naming the one connector and its /96", err, e)
+	}
+	issued, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Connector: e[0].Connector, Lifetime: time.Hour})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if issued.Prefix != e[0].Prefix || len(g.entries(t)) != 1 {
+		t.Fatalf("retry got %s with %d entries; want %s and one entry", issued.Prefix, len(g.entries(t)),
+			e[0].Prefix)
+	}
+}
+
+// TestCreateLostReply: a record written whose reply was lost is refused on
+// retry, and rotating it issues a credential for the same /96.
+func TestCreateLostReply(t *testing.T) {
+	n := 0
+	g, ref := faultyGrants(t, func() error {
+		if n++; n == 1 {
+			return errLost
+		}
+		return nil
+	})
+	ctx := context.Background()
+	if _, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Lifetime: time.Hour}); !errors.Is(err, errLost) {
+		t.Fatalf("create with a lost reply = %v", err)
+	}
+	e := g.entries(t)
+	_, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Connector: e[0].Connector, Lifetime: time.Hour})
+	if !apierrors.IsAlreadyExists(err) || !strings.Contains(err.Error(), "rotate") {
+		t.Fatalf("retry after a lost reply = %v; want AlreadyExists pointing at rotate", err)
+	}
+	issued, err := g.Rotate(ctx, e[0].Connector, time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if issued.Prefix != e[0].Prefix || issued.Credential == "" || len(g.entries(t)) != 1 {
+		t.Fatalf("rotate after a lost reply = %+v with %d entries; want %s", issued, len(g.entries(t)), e[0].Prefix)
+	}
+}
+
+// TestRetireUnrecorded: a connector whose record was never written can be
+// retired, which tombstones its /96 so no connector gets it again.
+func TestRetireUnrecorded(t *testing.T) {
+	g, ref := faultyGrants(t, failOnce())
+	ctx := context.Background()
+	if _, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Lifetime: time.Hour}); err == nil {
+		t.Fatal("create with a failed record write succeeded")
+	}
+	stranded := g.entries(t)[0]
+	if err := g.Retire(ctx, stranded.Connector); err != nil {
+		t.Fatalf("retire a connector without a record: %v", err)
+	}
+	if e := g.entries(t); len(e) != 1 || !e[0].Retired {
+		t.Fatalf("entries after retire %+v; want one tombstone", e)
+	}
+	if _, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Connector: stranded.Connector,
+		Lifetime: time.Hour}); err == nil {
+		t.Fatal("a retired connector was created again")
+	}
+	issued, err := g.Create(ctx, CreateRequest{VPC: ref, Cell: testCell, Lifetime: time.Hour})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if issued.Prefix == stranded.Prefix {
+		t.Fatalf("new connector got the retired %s", stranded.Prefix)
+	}
+	err = g.Retire(ctx, "33333333-3333-3333-3333-333333333333")
+	if !apierrors.IsNotFound(err) {
+		t.Fatalf("retire an unknown connector = %v, want NotFound", err)
+	}
+}

internal/edge/masque/authority/ledger.go

@@ -172,6 +172,26 @@ func (s Ledgers) Init(ctx context.Context, vpc VPCRef, cell string, connectors,
 	return nil
 }
 
+// holding returns every ledger with an entry for connector.
+func (s Ledgers) holding(ctx context.Context, connector string) ([]*Ledger, error) {
+	var cms corev1.ConfigMapList
+	if err := s.Client.List(ctx, &cms, client.InNamespace(s.Namespace),
+		client.MatchingLabels{LabelLedger: yes}); err != nil {
+		return nil, fmt.Errorf("list ledgers: %w", err)
+	}
+	var out []*Ledger
+	for _, cm := range cms.Items {
+		var l Ledger
+		if err := json.Unmarshal([]byte(cm.Data[ledgerKey]), &l); err != nil {
+			return nil, fmt.Errorf("decode ledger %s: %w", cm.Name, err)
+		}
+		if _, ok := l.Lookup(connector); ok {
+			out = append(out, &l)
+		}
+	}
+	return out, nil
+}
+
 // inNetworks reports whether p lies inside one of networks.
 func inNetworks(networks []cloudv1alpha1.Network, p netip.Prefix) bool {
 	for _, n := range networks {

internal/edge/masque/authority/ledger_test.go

@@ -45,12 +45,17 @@ func testScheme() *runtime.Scheme {
 	return s
 }
 
-func fakeLedgers(t *testing.T) (Ledgers, VPCRef) {
-	t.Helper()
-	vpc := &cloudv1alpha1.VPC{
+// testVPC returns the VPC the ledgers under test belong to.
+func testVPC() *cloudv1alpha1.VPC {
+	return &cloudv1alpha1.VPC{
 		ObjectMeta: metav1.ObjectMeta{Namespace: tenantNS, Name: vpcName, UID: "11111111-1111-1111-1111-111111111111"},
 		Spec:       cloudv1alpha1.VPCSpec{Networks: []cloudv1alpha1.Network{"fd20:a1b:2c3d::/48"}},
 	}
+}
+
+func fakeLedgers(t *testing.T) (Ledgers, VPCRef) {
+	t.Helper()
+	vpc := testVPC()
 	c := fake.NewClientBuilder().WithScheme(testScheme()).WithObjects(vpc).Build()
 	l := Ledgers{Client: c, Namespace: testNS, Now: time.Now}
 	ref := VPCRef{Namespace: tenantNS, Name: vpcName, UID: string(vpc.UID)}

tests/edgeconformance/edgeprobe/connectip.go

@@ -42,7 +42,8 @@ func connectIP(args []string) error {
 	token := fs.String("token", "", "bearer credential")
 	vpc := fs.String("vpc", "", "Galactic-VPC header value")
 	to := fs.String("to", "", "comma-separated echo destinations")
-	srcFlag := fs.String("src", "", "source address; default the first address after the assigned prefix's")
+	srcFlag := fs.String("src", "", "comma-separated source addresses, each sent to every destination; "+
+		"default the first address after the assigned prefix's")
 	id := fs.Int("id", os.Getpid()&0xffff, "ICMPv6 echo identifier")
 	size := fs.Int("size", 64, "IPv6 packet size in bytes")
 	interval := fs.Duration("interval", 50*time.Millisecond, "time between echo requests")
@@ -112,10 +113,15 @@ func connectIP(args []string) error {
 	}
 	prefix := assigned[0].IPPrefix
 	_ = out.Encode(cipEvent{T: boottime(), Assigned: prefix.String()})
-	src := prefix.Addr().Next()
+	srcs := []netip.Addr{prefix.Addr().Next()}
 	if *srcFlag != "" {
-		if src, err = netip.ParseAddr(*srcFlag); err != nil {
-			return fmt.Errorf("-src: %w", err)
+		srcs = nil
+		for _, a := range strings.Split(*srcFlag, ",") {
+			src, err := netip.ParseAddr(a)
+			if err != nil {
+				return fmt.Errorf("-src: %w", err)
+			}
+			srcs = append(srcs, src)
 		}
 	}
 
@@ -138,15 +144,17 @@ func connectIP(args []string) error {
 	sent := 0
 loop:
 	for time.Now().Before(deadline) {
-		for _, dst := range dsts {
-			pkt, err := echoRequest(src, dst, *id, sent, *size)
-			if err != nil {
-				return err
-			}
-			if _, err := conn.WritePacket(pkt); err != nil {
-				break loop
+		for _, src := range srcs {
+			for _, dst := range dsts {
+				pkt, err := echoRequest(src, dst, *id, sent, *size)
+				if err != nil {
+					return err
+				}
+				if _, err := conn.WritePacket(pkt); err != nil {
+					break loop
+				}
+				sent++
 			}
-			sent++
 		}
 		select {
 		case <-tick.C: