diff --git a/internal/api/api_test.go b/internal/api/api_test.go index e19b74f..d7b80bb 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -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). diff --git a/internal/api/handlers_patch_test.go b/internal/api/handlers_patch_test.go index 0716ffa..3fa16e0 100644 --- a/internal/api/handlers_patch_test.go +++ b/internal/api/handlers_patch_test.go @@ -1,6 +1,8 @@ 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) + } + }) +} diff --git a/internal/api/handlers_user.go b/internal/api/handlers_user.go index 20ab405..4295f47 100644 --- a/internal/api/handlers_user.go +++ b/internal/api/handlers_user.go @@ -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) { + 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 + } + if err != nil { 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) - 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) + } + } 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) diff --git a/internal/api/pgrepo.go b/internal/api/pgrepo.go index df42258..5a7064d 100644 --- a/internal/api/pgrepo.go +++ b/internal/api/pgrepo.go @@ -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); { - case errors.Is(err, sql.ErrNoRows): - return r, nil - case err != nil: - return r, err +// 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) + } } - return r, nil +} + +// 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 ResourceSpec{}, false, nil + case err != nil: + 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 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` diff --git a/internal/api/repo.go b/internal/api/repo.go index d9e6c80..279e58e 100644 --- a/internal/api/repo.go +++ b/internal/api/repo.go @@ -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 diff --git a/internal/pgint/resize_test.go b/internal/pgint/resize_test.go new file mode 100644 index 0000000..bee817e --- /dev/null +++ b/internal/pgint/resize_test.go @@ -0,0 +1,214 @@ +//go:build pgint + +package pgint + +import ( + "context" + "errors" + "testing" + + "felis.lolicon.best/internal/api" +) + +// ---- server resize (ResizeServer) -------------------------------------------------- + +func seedSized(t *testing.T, name, ownerID string, size api.ResourceSpec) { + t.Helper() + var owner any + if ownerID != "" { + owner = ownerID + } + mustExec(t, `INSERT INTO servers (name, cached_cpu_milli, cached_memory_mb, cached_storage_mb, owner_id) + VALUES ($1, $2, $3, $4, $5)`, name, size.CPUMilli, size.MemoryMB, size.StorageMB, owner) +} + +func cachedSize(t *testing.T, name string) api.ResourceSpec { + t.Helper() + var s api.ResourceSpec + if err := db.QueryRow(`SELECT cached_cpu_milli, cached_memory_mb, cached_storage_mb FROM servers WHERE name = $1`, + name).Scan(&s.CPUMilli, &s.MemoryMB, &s.StorageMB); err != nil { + t.Fatalf("cached size of %s: %v", name, err) + } + return s +} + +// A resize that grows CPU or memory must fit every cap with the server's whole size +// counted, its storage and itself among the owner's servers included; one that does +// not writes nothing. A resize that grows neither is written whatever the caps say, +// and so is any resize of a server nobody owns. +func TestResizeServerQuota(t *testing.T) { + ctx := context.Background() + n := func(v int) *int { return &v } + u := newUser(t, "user", "rsz") + sfx := suffix(t) + a, b := "rsa-"+sfx, "rsb-"+sfx + seedSized(t, a, u.ID, api.ResourceSpec{CPUMilli: 1000, MemoryMB: 2048, StorageMB: 400}) + seedSized(t, b, u.ID, api.ResourceSpec{CPUMilli: 1000, MemoryMB: 2048, StorageMB: 400}) + if _, err := repo.SetQuotas(ctx, u.ID, api.QuotaInput{MaxServers: n(2), MaxCPUMilli: n(3000), + MaxMemoryMB: n(6144), MaxStorageGB: n(1)}, "pgint"); err != nil { + t.Fatalf("SetQuotas: %v", err) + } + + for _, c := range []struct { + name string + cpu, mem int + refused bool + after api.ResourceSpec + }{ + {"cpu grows to the cap", 2000, 2048, false, api.ResourceSpec{CPUMilli: 2000, MemoryMB: 2048, StorageMB: 400}}, + {"cpu grows past the cap", 2001, 2048, true, api.ResourceSpec{CPUMilli: 2000, MemoryMB: 2048, StorageMB: 400}}, + {"memory grows past the cap", 2000, 4097, true, api.ResourceSpec{CPUMilli: 2000, MemoryMB: 2048, StorageMB: 400}}, + {"memory grows to the cap as cpu shrinks", 1500, 4096, false, api.ResourceSpec{CPUMilli: 1500, MemoryMB: 4096, StorageMB: 400}}, + } { + prevWant := cachedSize(t, a) + prev, err := repo.ResizeServer(ctx, a, c.cpu, c.mem) + switch { + case c.refused && !errors.Is(err, api.ErrQuotaExceeded): + t.Fatalf("%s: err = %v, want ErrQuotaExceeded", c.name, err) + case !c.refused && (err != nil || prev != prevWant): + t.Fatalf("%s: prev = %+v, err = %v; want %+v, nil", c.name, prev, err, prevWant) + } + if got := cachedSize(t, a); got != c.after { + t.Fatalf("%s: cache = %+v, want %+v", c.name, got, c.after) + } + } + + // Storage over its cap (an admin lowered it, or the other server grew): growth + // is refused though it fits CPU and memory, and a shrink still goes through. + mustExec(t, `UPDATE servers SET cached_storage_mb = 700 WHERE name = $1`, b) + if _, err := repo.ResizeServer(ctx, a, 1600, 4096); !errors.Is(err, api.ErrQuotaExceeded) { + t.Fatalf("growth with storage over its cap: err = %v, want ErrQuotaExceeded", err) + } + if _, err := repo.ResizeServer(ctx, a, 1000, 2048); err != nil { + t.Fatalf("shrink with storage over its cap: %v", err) + } + if got, want := cachedSize(t, a), (api.ResourceSpec{CPUMilli: 1000, MemoryMB: 2048, StorageMB: 400}); got != want { + t.Fatalf("cache after the shrink = %+v, want %+v", got, want) + } + + free := "rsf-" + sfx + seedSized(t, free, "", api.ResourceSpec{CPUMilli: 500, MemoryMB: 1024, StorageMB: 400}) + if prev, err := repo.ResizeServer(ctx, free, 8000, 16384); err != nil || prev.CPUMilli != 500 { + t.Fatalf("resize of an unowned server = %+v, %v; want prev cpu 500, nil", prev, err) + } + if got, want := cachedSize(t, free), (api.ResourceSpec{CPUMilli: 8000, MemoryMB: 16384, StorageMB: 400}); got != want { + t.Fatalf("unowned cache = %+v, want %+v", got, want) + } + + if prev, err := repo.ResizeServer(ctx, "rsnone-"+sfx, 1000, 1024); err != nil || prev != (api.ResourceSpec{}) { + t.Fatalf("resize of no server = %+v, %v; want zeroes, nil", prev, err) + } + if rowExists(t, `SELECT 1 FROM servers WHERE name = $1`, "rsnone-"+sfx) { + t.Fatal("a resize of no server made a row") + } +} + +// A resize decides in its owner's claim lane: a server the owner claims while the +// resize waits is counted, and here tips the owner over the CPU cap. +func TestResizeServerWaitsForClaims(t *testing.T) { + ctx := context.Background() + n := func(v int) *int { return &v } + u := newUser(t, "user", "rszw") + sfx := suffix(t) + a, claimed := "rswa-"+sfx, "rswc-"+sfx + seedSized(t, a, u.ID, api.ResourceSpec{CPUMilli: 1000, MemoryMB: 1024, StorageMB: 1}) + seedSized(t, claimed, "", api.ResourceSpec{CPUMilli: 1000, MemoryMB: 1024, StorageMB: 1}) + if _, err := repo.SetQuotas(ctx, u.ID, api.QuotaInput{MaxCPUMilli: n(2000)}, "pgint"); err != nil { + t.Fatalf("SetQuotas: %v", err) + } + + hold := holdLane(t, u.ID) + defer hold.Rollback() //nolint:errcheck // no-op after commit + resized := make(chan error, 1) + go func() { + _, err := repo.ResizeServer(ctx, a, 1500, 1024) + resized <- err + }() + waitForLockWait(t, "SELECT pg_advisory_xact_lock") + if _, err := hold.Exec(`UPDATE servers SET owner_id = $2 WHERE name = $1`, claimed, u.ID); err != nil { + t.Fatalf("claim: %v", err) + } + if err := hold.Commit(); err != nil { + t.Fatalf("claim commit: %v", err) + } + if err := <-resized; !errors.Is(err, api.ErrQuotaExceeded) { + t.Fatalf("resize behind a claim = %v, want ErrQuotaExceeded", err) + } + if got := cachedSize(t, a); got.CPUMilli != 1000 { + t.Fatalf("cache = %+v, want cpu still 1000", got) + } +} + +// A server that changes hands while its resize waits on the old owner's lane is +// judged against the new owner's caps, under the new owner's lane. +func TestResizeServerFollowsNewOwner(t *testing.T) { + ctx := context.Background() + n := func(v int) *int { return &v } + was := newUser(t, "user", "rszo-was") + now := newUser(t, "user", "rszo-now") + a := "rsoa-" + suffix(t) + seedSized(t, a, was.ID, api.ResourceSpec{CPUMilli: 1000, MemoryMB: 1024, StorageMB: 1}) + if _, err := repo.SetQuotas(ctx, now.ID, api.QuotaInput{MaxCPUMilli: n(1000)}, "pgint"); err != nil { + t.Fatalf("SetQuotas: %v", err) + } + + hold := holdLane(t, was.ID) + defer hold.Rollback() //nolint:errcheck // no-op after commit + resized := make(chan error, 1) + go func() { + _, err := repo.ResizeServer(ctx, a, 1500, 1024) + resized <- err + }() + waitForLockWait(t, "SELECT pg_advisory_xact_lock") + mustExec(t, `UPDATE servers SET owner_id = $2 WHERE name = $1`, a, now.ID) + if err := hold.Commit(); err != nil { + t.Fatalf("release lane: %v", err) + } + if err := <-resized; !errors.Is(err, api.ErrQuotaExceeded) { + t.Fatalf("resize of a server that changed hands = %v, want ErrQuotaExceeded from the new owner's cap", err) + } + if got := cachedSize(t, a); got.CPUMilli != 1000 { + t.Fatalf("cache = %+v, want cpu still 1000", got) + } +} + +// A server nobody owns has no lane to take, so its row lock is what orders a resize +// against a claim: a resize that reaches the row while a claim holds it waits, finds +// the claimer's name on it, and is judged against the claimer's caps. +func TestResizeServerWaitsForAClaimOfAFreeServer(t *testing.T) { + ctx := context.Background() + n := func(v int) *int { return &v } + claimer := newUser(t, "user", "rszf") + free := "rsfc-" + suffix(t) + seedSized(t, free, "", api.ResourceSpec{CPUMilli: 1000, MemoryMB: 1024, StorageMB: 1}) + if _, err := repo.SetQuotas(ctx, claimer.ID, api.QuotaInput{MaxCPUMilli: n(1000)}, "pgint"); err != nil { + t.Fatalf("SetQuotas: %v", err) + } + + hold, err := db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("begin: %v", err) + } + defer hold.Rollback() //nolint:errcheck // no-op after commit + if _, err := hold.Exec(`SELECT 1 FROM servers WHERE name = $1 FOR UPDATE`, free); err != nil { + t.Fatalf("hold server: %v", err) + } + if _, err := hold.Exec(`UPDATE servers SET owner_id = $2 WHERE name = $1`, free, claimer.ID); err != nil { + t.Fatalf("claim: %v", err) + } + resized := make(chan error, 1) + go func() { + _, err := repo.ResizeServer(ctx, free, 1500, 1024) + resized <- err + }() + waitForLockWait(t, "SELECT owner_id, cached_cpu_milli") + if err := hold.Commit(); err != nil { + t.Fatalf("claim commit: %v", err) + } + if err := <-resized; !errors.Is(err, api.ErrQuotaExceeded) { + t.Fatalf("resize behind a claim of a free server = %v, want ErrQuotaExceeded", err) + } + if got := cachedSize(t, free); got.CPUMilli != 1000 { + t.Fatalf("cache = %+v, want cpu still 1000", got) + } +}