fix(api): 调整资源先在 owner 的认领锁内核配额并写缓存再 patch 集群，集群拒绝时缓存回到集群的大小 (2c5a85a0) · Commits · FelisMC / Felis · GitLab

internal/api/api_test.go

+39 −6

Changes for internal/api/api_test.go: 39 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -51,11 +51,13 @@ type fakeRepo struct {
		// finds too small for the servers moving in; it refuses only a source that owns one.
		migrateQuotaRefuse map[string]bool
		// serverResources / resourceUpdates mirror the cached resource columns:
		// ServerResources is what the resize path reads (to preserve storage), and
		// UpdateServerResources records the write for assertions.
		// serverResources is what the cache held (ResizeServer keeps its storage), and
		// UpdateServerResources and ResizeServer record their writes for assertions.
		serverResources map[string]ResourceSpec
		resourceUpdates map[string]ResourceSpec
		// quotaChecked records the size every QuotaCheck was asked about, in order.
		resizeErr error // ResizeServer fails with it (a store outage)
		// quotaChecked records the size every QuotaCheck, and every ResizeServer that
		// grows an owned server, was asked about, in order.
		quotaChecked []ResourceSpec
		audits []AuditEntry
		failAudit error // Audit fails with it (a store outage)
		@@ -342,14 +344,36 @@ func (f *fakeRepo) QuotaCheck(_ context.Context, userID string, _ string, incomi
		return f.QuotaAvailable(context.TODO(), userID)
		}

		func (f *fakeRepo) UpdateServerResources(_ context.Context, name string, cpu, mem, stor int) error {
		func (f *fakeRepo) UpdateServerResources(ctx context.Context, name string, cpu, mem, stor int) error {
		if err := ctx.Err(); err != nil {
		return err // as the database driver refuses a done context
		}
		f.resourceUpdates[name] = ResourceSpec{CPUMilli: cpu, MemoryMB: mem, StorageMB: stor}
		return nil
		}

		func (f *fakeRepo) ServerResources(_ context.Context, name string) (ResourceSpec, error) {
		return f.serverResources[name], nil
		// ResizeServer mirrors PGRepo.ResizeServer: an owned server that grows is checked
		// with its whole size, storage taken from the cache, and a refusal writes nothing.
		func (f *fakeRepo) ResizeServer(_ context.Context, name string, cpu, mem int) (ResourceSpec, error) {
		if f.resizeErr != nil {
		return ResourceSpec{}, f.resizeErr
		}
		rec, ok := f.byName[name]
		if !ok {
		return ResourceSpec{}, nil
		}
		prev := f.serverResources[name]
		next := ResourceSpec{CPUMilli: cpu, MemoryMB: mem, StorageMB: prev.StorageMB}
		if rec.OwnerID != "" && (cpu > prev.CPUMilli \|\| mem > prev.MemoryMB) {
		f.quotaChecked = append(f.quotaChecked, next)
		if !f.quota[rec.OwnerID] {
		return ResourceSpec{}, ErrQuotaExceeded
		}
		}
		f.resourceUpdates[name] = next
		return prev, nil
		}

		func (f *fakeRepo) CreateLinkCode(_ context.Context, code, mcUUID, authSource string, expiresAt time.Time) error {
		f.linkCodes[code] = fakeLinkCode{mcUUID: mcUUID, authSource: authSource, expiresAt: expiresAt}
		return nil
		@@ -1806,6 +1830,8 @@ type fakeCluster struct {
		orphanWorld map[string]bool // names with a world volume but no server (CR deleted by hand)
		createErr error
		pingErr error
		patchErr error // PatchServerSpec fails with it after finding the server
		goneOnPatch bool // PatchServerSpec finds the server deleted since it was read
		// maintErr / wakeErr: what AcquireMaintenance / SetDesiredState(Running)
		// return for a server (the world-volume lock, internal/maintenance).
		maintErr map[string]error
		@@ -1903,6 +1929,13 @@ func (c *fakeCluster) PatchServerSpec(_ context.Context, n string, p ServerSpecP
		if !ok {
		return ErrNotFound
		}
		if c.goneOnPatch {
		delete(c.byName, n)
		return ErrNotFound
		}
		if c.patchErr != nil {
		return c.patchErr
		}
		c.patched[n] = p
		// Apply only the fields the lifecycle view exposes, so a follow-up read sees
		// the mutation (mirrors the real merge patch touching only non-nil fields).

internal/api/handlers_patch_test.go

+77 −0

Changes for internal/api/handlers_patch_test.go: 77 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		package api

		import (
		"context"
		"errors"
		"net/http"
		"net/http/httptest"
		"strings"
		@@ -474,3 +476,78 @@ func TestPatchServerOverQuotaMayShrink(t *testing.T) {
		}
		}
		}

		// A resize is written to the resource cache, the figures the owner's quota sums
		// read, before the cluster sees it. The write used to come after the patch with its
		// error dropped, so a failed write left the owner's new size uncounted. When the
		// cluster then refuses the patch, the cache is put back: to what the cluster
		// reports, or, when the server is gone from it, to what the cache held.
		func TestPatchServerResizeCache(t *testing.T) {
		const body = `{"resources":{"cpu":"2"}}`
		before := ResourceSpec{CPUMilli: 750, MemoryMB: 2048, StorageMB: 5120}
		mk := func() (API, fakeRepo, *fakeCluster) {
		api, repo, cl, _ := newPatchAPI()
		seedResources(cl)
		cl.byName["survival"].StorageSize = "10Gi"
		repo.serverResources["survival"] = before
		return api, repo, cl
		}

		t.Run("a cache that cannot be written stops the patch", func(t *testing.T) {
		api, repo, cl := mk()
		repo.resizeErr = errors.New("database unreachable")
		if w := patchSurvival(api, body); w.Code != http.StatusInternalServerError {
		t.Fatalf("code = %d body %s, want 500", w.Code, w.Body.String())
		}
		if _, ok := cl.patched["survival"]; ok \|\| len(repo.audits) != 0 {
		t.Fatalf("patched=%+v audits=%+v, want neither", cl.patched, repo.audits)
		}
		})

		t.Run("the patch lands with the new size cached", func(t *testing.T) {
		api, repo, cl := mk()
		if w := patchSurvival(api, body); w.Code != http.StatusOK {
		t.Fatalf("code = %d body %s, want 200", w.Code, w.Body.String())
		}
		if _, ok := cl.patched["survival"]; !ok {
		t.Fatal("the patch did not reach the cluster")
		}
		if got, want := repo.resourceUpdates["survival"], (ResourceSpec{CPUMilli: 2000, MemoryMB: 4096, StorageMB: 5120}); got != want {
		t.Fatalf("resource cache = %+v, want %+v", got, want)
		}
		})

		t.Run("the cluster refuses: the cache takes the cluster's size", func(t *testing.T) {
		api, repo, cl := mk()
		cl.patchErr = errors.New("apiserver unavailable")
		if w := patchSurvival(api, body); w.Code != http.StatusInternalServerError {
		t.Fatalf("code = %d body %s, want 500", w.Code, w.Body.String())
		}
		if got, want := repo.resourceUpdates["survival"], (ResourceSpec{CPUMilli: 1000, MemoryMB: 4096, StorageMB: 10240}); got != want {
		t.Fatalf("resource cache = %+v, want the cluster's %+v", got, want)
		}
		})

		t.Run("the server is gone: the cache takes back what it held", func(t *testing.T) {
		api, repo, cl := mk()
		cl.goneOnPatch = true
		if w := patchSurvival(api, body); w.Code != http.StatusNotFound \|\| decodeErr(t, w) != "not_found" {
		t.Fatalf("code = %d body %s, want 404 not_found", w.Code, w.Body.String())
		}
		if got := repo.resourceUpdates["survival"]; got != before {
		t.Fatalf("resource cache = %+v, want %+v", got, before)
		}
		})

		t.Run("a client that hangs up still gets the cache put back", func(t *testing.T) {
		api, repo, cl := mk()
		cl.patchErr = context.Canceled
		ctx, cancel := context.WithCancel(context.Background())
		cancel()
		r := httptest.NewRequest("PATCH", "/api/v1/servers/survival", strings.NewReader(body)).WithContext(ctx)
		api.ExternalHandler().ServeHTTP(httptest.NewRecorder(), r)
		if got, want := repo.resourceUpdates["survival"], (ResourceSpec{CPUMilli: 1000, MemoryMB: 4096, StorageMB: 10240}); got != want {
		t.Fatalf("resource cache = %+v, want the cluster's %+v", got, want)
		}
		})
		}

internal/api/handlers_user.go

+20 −30

Changes for internal/api/handlers_user.go: 20 added lines, 30 removed lines.
Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@ import (
		"context"
		"errors"
		"fmt"
		"log"
		"net/http"
		"strings"

		@@ -1046,51 +1047,40 @@ func (a API) handlePatchServer(w http.ResponseWriter, r http.Request) {
		resUpdated = true
		}

		// Resource-cache consistency + quota enforcement (spec §9.3 / §22): every
		// resource-mutating patch must update the cached columns so QuotaCheck
		// can aggregate per-owner usage without cross-system CRD reads. For OWNED
		// servers the owner's cumulative usage must also stay within their quota caps.
		// Resource-cache consistency + quota enforcement (spec §9.3 / §22): the cached
		// columns are all the per-owner quota sums read, so the new size is written
		// there before the cluster sees it, in the owner's claim lane and only if it
		// fits (ResizeServer). Only growth is held to the caps: a change that grows
		// neither CPU nor memory cannot push the owner past one, and it is how an admin
		// brings a server back under a cap lowered below what the owner already uses.
		if resUpdated {
		newCPU := quantityToMilli(newResources.Limits[corev1.ResourceCPU])
		newMemMB := quantityToMB(newResources.Limits[corev1.ResourceMemory])

		rec, err := a.Repo.ServerByName(r.Context(), name)
		if err != nil && !errors.Is(err, ErrNotFound) {
		writeError(w, r, err)
		return
		}
		var cur ResourceSpec
		if rec != nil {
		if cur, err = a.Repo.ServerResources(r.Context(), name); err != nil {
		writeError(w, r, err)
		prev, err := a.Repo.ResizeServer(r.Context(), name, newCPU, newMemMB)
		if errors.Is(err, ErrQuotaExceeded) {
		writeError(w, r, newError(http.StatusForbidden, "quota_exceeded",
		"this change would exceed the server owner's resource quota"))
		return
		}
		}
		// Only growth is held to the caps. A change that grows neither CPU nor memory
		// cannot push the owner past one, and it is how an admin brings a server back
		// under a cap lowered below what the owner already uses.
		if rec != nil && rec.OwnerID != "" && (newCPU > cur.CPUMilli \|\| newMemMB > cur.MemoryMB) {
		ok, err := a.Repo.QuotaCheck(r.Context(), rec.OwnerID, name,
		ResourceSpec{CPUMilli: newCPU, MemoryMB: newMemMB})
		if err != nil {
		writeError(w, r, err)
		return
		}
		if !ok {
		writeError(w, r, newError(http.StatusForbidden, "quota_exceeded",
		"this change would exceed the server owner's resource quota"))
		return
		if err := a.Cluster.PatchServerSpec(r.Context(), name, patch); err != nil {
		// The cache now holds a size the cluster may not: it goes back to the size
		// the cluster reports, or, when that cannot be read either, to the one it
		// held before.
		ctx := context.WithoutCancel(r.Context())
		if _, rerr := a.claimResources(ctx, name); rerr != nil {
		if rerr := a.Repo.UpdateServerResources(ctx, name, prev.CPUMilli, prev.MemoryMB, prev.StorageMB); rerr != nil {
		log.Printf("api: resize of %s refused by the cluster, resource cache left at the new size (request_id=%s): %v",
		name, requestIDFromContext(ctx), rerr)
		}
		}

		if err := a.Cluster.PatchServerSpec(r.Context(), name, patch); err != nil {
		a.writeLookupError(w, r, err)
		return
		}
		// A resource patch cannot change storage, so its cached contribution must
		// be preserved: passing 0 would silently zero the storage dimension of the
		// owner's four-cap aggregate (the cached columns are its only input).
		_ = a.Repo.UpdateServerResources(r.Context(), name, newCPU, newMemMB, cur.StorageMB)
		} else {
		if err := a.Cluster.PatchServerSpec(r.Context(), name, patch); err != nil {
		a.writeLookupError(w, r, err)

internal/api/pgrepo.go

+83 −23

Changes for internal/api/pgrepo.go: 83 added lines, 23 removed lines.
Original line number	Diff line number	Diff line
		@@ -371,13 +371,16 @@ func (p *PGRepo) QuotaAvailable(ctx context.Context, userID string) (bool, error
		// doesn't exist yet). Four dimensions are checked: server count, CPU millicores,
		// memory MB, and storage MB. A NULL or missing quota row/column means unlimited
		// for that dimension. As a standalone read it is advisory — it backs the
		// handler's fast-path 403 — while the AUTHORITATIVE gate for claims is the one
		// ClaimServer re-runs atomically; the resize path (server PATCH) keeps this
		// advisory shape because its write goes through the Kubernetes API, not this
		// transaction.
		// handler's fast-path 403 — while the AUTHORITATIVE gates are the ones
		// ClaimServer and ResizeServer run in the owner's claim lane.
		func (p *PGRepo) QuotaCheck(ctx context.Context, userID string, excludeName string, incoming ResourceSpec) (bool, error) {
		return quotaFits(ctx, p.db, userID, excludeName, incoming)
		}

		// quotaFits is QuotaCheck over either the pool or a transaction.
		func quotaFits(ctx context.Context, q rowQuerier, userID, excludeName string, incoming ResourceSpec) (bool, error) {
		var maxServers, maxCPU, maxMem, maxStor sql.NullInt64
		switch err := p.db.QueryRowContext(ctx,
		switch err := q.QueryRowContext(ctx,
		`SELECT max_servers, max_cpu_milli, max_memory_mb, max_storage_gb
		FROM quotas WHERE user_id = $1`, userID).Scan(
		&maxServers, &maxCPU, &maxMem, &maxStor); {
		@@ -388,20 +391,19 @@ func (p *PGRepo) QuotaCheck(ctx context.Context, userID string, excludeName stri
		}

		var count, cpuSum, memSum, storSum int64
		switch err := p.db.QueryRowContext(ctx,
		if err := q.QueryRowContext(ctx,
		`SELECT COUNT(*), COALESCE(SUM(cached_cpu_milli), 0), COALESCE(SUM(cached_memory_mb), 0), COALESCE(SUM(cached_storage_mb), 0)
		FROM servers WHERE owner_id = $1 AND deleted_at IS NULL AND name != $2`,
		userID, excludeName).Scan(&count, &cpuSum, &memSum, &storSum); {
		case err != nil:
		userID, excludeName).Scan(&count, &cpuSum, &memSum, &storSum); err != nil {
		return false, err
		}

		return quotaAllows(maxServers, maxCPU, maxMem, maxStor, count, cpuSum, memSum, storSum, 1, incoming), nil
		}

		// UpdateServerResources updates the resource cache for a server after a spec
		// mutation (spec §7 PATCH). The per-owner aggregate used by QuotaCheck is a
		// SQL SUM over the cached columns, so every mutation must write through here.
		// UpdateServerResources overwrites a server's resource cache. The per-owner
		// aggregate used by QuotaCheck is a SQL SUM over the cached columns, so every
		// size the cluster takes must reach them, through here or ResizeServer.
		func (p *PGRepo) UpdateServerResources(ctx context.Context, name string, cpuMilli, memoryMB, storageMB int) error {
		_, err := p.db.ExecContext(ctx,
		`UPDATE servers SET cached_cpu_milli = $2, cached_memory_mb = $3, cached_storage_mb = $4 WHERE name = $1 AND deleted_at IS NULL`,
		@@ -409,18 +411,76 @@ func (p *PGRepo) UpdateServerResources(ctx context.Context, name string, cpuMill
		return err
		}

		// ServerResources returns the cached resource spec for a server.
		func (p *PGRepo) ServerResources(ctx context.Context, name string) (ResourceSpec, error) {
		var r ResourceSpec
		switch err := p.db.QueryRowContext(ctx,
		`SELECT cached_cpu_milli, cached_memory_mb, cached_storage_mb FROM servers WHERE name = $1 AND deleted_at IS NULL`,
		name).Scan(&r.CPUMilli, &r.MemoryMB, &r.StorageMB); {
		// ResizeServer gates and records a CPU and memory resize before the cluster sees
		// it. The owner is only known from the row, and the lane must be taken before the
		// row lock (a redeem holds lanes while it moves rows), so a server that changes
		// hands in between is read again in a fresh transaction under its new owner's lane.
		func (p *PGRepo) ResizeServer(ctx context.Context, name string, cpuMilli, memoryMB int) (ResourceSpec, error) {
		for attempt := 1; ; attempt++ {
		prev, moved, err := p.resizeServer(ctx, name, cpuMilli, memoryMB)
		if !moved {
		return prev, err
		}
		if attempt == 3 {
		return ResourceSpec{}, fmt.Errorf("resize %s: its owner changed three times while it waited", name)
		}
		}
		}

		// resizeServer is one attempt of ResizeServer. moved reports that the owner it
		// locked for is no longer the row's, and nothing was written.
		func (p *PGRepo) resizeServer(ctx context.Context, name string, cpuMilli, memoryMB int) (prev ResourceSpec, moved bool, err error) {
		tx, err := p.db.BeginTx(ctx, nil)
		if err != nil {
		return ResourceSpec{}, false, err
		}
		defer tx.Rollback() //nolint:errcheck // no-op after commit

		var owner sql.NullString
		switch err := tx.QueryRowContext(ctx,
		`SELECT owner_id FROM servers WHERE name = $1 AND deleted_at IS NULL`, name).Scan(&owner); {
		case errors.Is(err, sql.ErrNoRows):
		return r, nil
		return ResourceSpec{}, false, nil
		case err != nil:
		return r, err
		return ResourceSpec{}, false, err
		}
		if owner.Valid {
		if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, owner.String); err != nil {
		return ResourceSpec{}, false, err
		}
		}
		var locked sql.NullString
		switch err := tx.QueryRowContext(ctx,
		`SELECT owner_id, cached_cpu_milli, cached_memory_mb, cached_storage_mb
		FROM servers WHERE name = $1 AND deleted_at IS NULL FOR UPDATE`,
		name).Scan(&locked, &prev.CPUMilli, &prev.MemoryMB, &prev.StorageMB); {
		case errors.Is(err, sql.ErrNoRows):
		return ResourceSpec{}, false, nil
		case err != nil:
		return ResourceSpec{}, false, err
		}
		if locked != owner {
		return ResourceSpec{}, true, nil
		}

		// A server nobody owns is checked against the empty owner, who has no caps:
		// every quotas row belongs to a user.
		if cpuMilli > prev.CPUMilli \|\| memoryMB > prev.MemoryMB {
		ok, err := quotaFits(ctx, tx, owner.String, name,
		ResourceSpec{CPUMilli: cpuMilli, MemoryMB: memoryMB, StorageMB: prev.StorageMB})
		if err != nil {
		return ResourceSpec{}, false, err
		}
		if !ok {
		return ResourceSpec{}, false, ErrQuotaExceeded
		}
		}
		if _, err := tx.ExecContext(ctx,
		`UPDATE servers SET cached_cpu_milli = $2, cached_memory_mb = $3 WHERE name = $1`,
		name, cpuMilli, memoryMB); err != nil {
		return ResourceSpec{}, false, err
		}
		return r, nil
		return prev, false, tx.Commit()
		}

		// ClaimServer performs the atomic ownership transfer (spec §9.3). A missing
		@@ -2696,15 +2756,15 @@ func (p *PGRepo) OTPLockedUntil(ctx context.Context, userID, purpose string, now
		return otpLockedUntil(ctx, p.db, userID, purpose, now, false)
		}

		// otpQuerier is the read half shared by sql.DB and sql.Tx.
		type otpQuerier interface {
		// rowQuerier is the read half shared by sql.DB and sql.Tx.
		type rowQuerier interface {
		QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
		}

		// otpLockedUntil returns when the (user, purpose) lock ends, or zero when the
		// budget is not spent in the current window. forUpdate takes the row lock so a
		// redeem serialises its check with its own charge.
		func otpLockedUntil(ctx context.Context, q otpQuerier, userID, purpose string, now time.Time, forUpdate bool) (time.Time, error) {
		func otpLockedUntil(ctx context.Context, q rowQuerier, userID, purpose string, now time.Time, forUpdate bool) (time.Time, error) {
		query := `SELECT window_start, failures FROM otp_failure_windows WHERE user_id = $1 AND purpose = $2`
		if forUpdate {
		query += ` FOR UPDATE`

internal/api/repo.go

+12 −6

Changes for internal/api/repo.go: 12 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -432,13 +432,19 @@ type Repo interface {
		// ErrConflict, and changes nothing, if the subdomain is already bound to a
		// different server.
		SeedServer(ctx context.Context, name, subdomain string, cpuMilli, memoryMB, storageMB int) error
		// UpdateServerResources updates the resource cache columns for a server
		// after a spec mutation (spec §7 PATCH), so the per-owner aggregate stays in
		// sync.
		// UpdateServerResources overwrites a server's resource cache with the size the
		// cluster holds (a claim writes it through; a resize the cluster refused is put
		// back with it), so the per-owner aggregate stays in sync.
		UpdateServerResources(ctx context.Context, name string, cpuMilli, memoryMB, storageMB int) error
		// ServerResources returns the cached resource spec for a server, or zeroes
		// when the row does not exist or has been cleared.
		ServerResources(ctx context.Context, name string) (ResourceSpec, error)
		// ResizeServer writes a server's new CPU and memory to its resource cache,
		// keeping its storage, in the owner's claim lane (the lock ClaimServer and
		// RedeemMigration take): resizes and claims of one owner's servers queue, and
		// each counts the sizes the others wrote. A resize that grows CPU or memory
		// must first fit every cap with the server's whole size counted; one that does
		// not returns ErrQuotaExceeded and writes nothing. prev is what the cache held.
		// A server with no owner has no caps to fit, and one with no live row writes
		// nothing and returns zeroes.
		ResizeServer(ctx context.Context, name string, cpuMilli, memoryMB int) (prev ResourceSpec, err error)
		// Audit appends one audit row.
		Audit(ctx context.Context, e AuditEntry) error