fix(master): let the growth initiator wait instead of shedding itself (#10202)

* fix(master): let the growth initiator wait for the growth it triggered The growth-in-flight shed also fired on the request that initiated the growth: it sets the pending flag right before the shed check, so a cold-start assign enqueued growth and immediately failed itself with "volume growth in progress". With no concurrent assigns around to pick up the freshly grown volume, a single writer against an empty cluster never completes a write despite ample free space. Claim the pending flag with a compare-and-swap so exactly one request becomes the initiator, triggering growth at most once, and let it wait for that growth to land. Everyone else still sheds retryably instead of pinning a goroutine: followers behind an in-flight growth, an initiator whose growth concluded without yielding a writable volume, and an initiator whose growth outlives the 10s wait budget, which previously surfaced a non-retryable error (gRPC Unknown, HTTP 406) even though a retry would have succeeded moments later. * fix(master): stop assign waits when the request is cancelled The assign retry loops slept through client cancellation, keeping a goroutine spinning for the rest of the 10s budget after the caller had gone; StreamAssign also ran assigns on a background context detached from the stream. Wait on the request context and pass the stream context through. * topology: drop the unconditional grow-request setter Growth is only claimed through AddGrowRequestIfAbsent's compare-and-swap now; keeping the raw Store(true) around invites the check-then-set race back. * test: cover cold-start first write with a real cluster Boot a fresh master plus three empty volume servers and require the very first assign - HTTP and gRPC, each on a cold volume layout, no client retries - to complete a write. The assign that triggers volume growth must wait for it rather than answering "volume growth in progress"; unit tests stub the topology, so only a real cluster exercises the assign-grow-wait path end to end.
2026-11-12 14:58:44 +01:00 · 2026-07-02 15:13:46 -07:00
1 parent c5240944e4
commit 6206f60032
@@ -0,0 +1,119 @@
+package master_cold_start
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+// The very first write against a fresh cluster must succeed in one shot: the
+// assign that triggers volume growth waits for it instead of shedding itself
+// with "volume growth in progress". No client retries — a single request per
+// assign, like a plain HTTP client.
+func TestColdStartFirstWrite(t *testing.T) {
+	c := StartCluster(t, 3)
+
+	// Each subtest uses its own replication so its volume layout starts cold.
+	t.Run("http_assign_and_write", func(t *testing.T) {
+		client := &http.Client{Timeout: 15 * time.Second}
+		resp, err := client.Get(c.MasterURL() + "/dir/assign?replication=002")
+		if err != nil {
+			t.Fatalf("dir/assign: %v", err)
+		}
+		body, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			t.Fatalf("dir/assign returned %d: %s", resp.StatusCode, body)
+		}
+		var assign struct {
+			Fid   string `json:"fid"`
+			Url   string `json:"url"`
+			Error string `json:"error"`
+		}
+		if err := json.Unmarshal(body, &assign); err != nil {
+			t.Fatalf("parse assign response %q: %v", body, err)
+		}
+		if assign.Error != "" || assign.Fid == "" {
+			t.Fatalf("assign failed: %s", body)
+		}
+
+		content := []byte("cold start write")
+		if err := uploadAndReadBack(client, assign.Url, assign.Fid, content); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("grpc_assign", func(t *testing.T) {
+		grpcDialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+		err := pb.WithMasterClient(context.Background(), false, pb.ServerAddress(c.MasterAddress()), grpcDialOption, false,
+			func(client master_pb.SeaweedClient) error {
+				ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+				defer cancel()
+				resp, err := client.Assign(ctx, &master_pb.AssignRequest{Count: 1, Replication: "001"})
+				if err != nil {
+					return err
+				}
+				if resp.Error != "" || resp.Fid == "" {
+					return fmt.Errorf("assign failed: %+v", resp)
+				}
+				return nil
+			})
+		if err != nil {
+			t.Fatalf("single grpc assign on cold cluster: %v", err)
+		}
+	})
+}
+
+// uploadAndReadBack writes content to the assigned fid and reads it back.
+func uploadAndReadBack(client *http.Client, volumeUrl, fid string, content []byte) error {
+	var buf bytes.Buffer
+	writer := multipart.NewWriter(&buf)
+	part, err := writer.CreateFormFile("file", "cold_start.txt")
+	if err != nil {
+		return err
+	}
+	part.Write(content)
+	writer.Close()
+
+	target := fmt.Sprintf("http://%s/%s", volumeUrl, fid)
+	req, err := http.NewRequest(http.MethodPost, target, &buf)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", writer.FormDataContentType())
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("upload to %s: %w", target, err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("upload to %s returned %d: %s", target, resp.StatusCode, body)
+	}
+
+	resp, err = client.Get(target)
+	if err != nil {
+		return fmt.Errorf("read back %s: %w", target, err)
+	}
+	got, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("read back %s returned %d", target, resp.StatusCode)
+	}
+	if !bytes.Equal(got, content) {
+		return fmt.Errorf("read back %s: got %q, want %q", target, got, content)
+	}
+	return nil
+}
@@ -0,0 +1,297 @@
+// Package master_cold_start boots a fresh master plus empty volume servers
+// and exercises the very first write against the cluster. The first assign
+// triggers volume growth server-side; it must succeed within a single
+// request, without client retries.
+package master_cold_start
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+)
+
+const (
+	waitTimeout = 30 * time.Second
+	waitTick    = 200 * time.Millisecond
+)
+
+type process struct {
+	name    string
+	cmd     *exec.Cmd
+	logFile string
+}
+
+// Cluster is one fresh master plus n empty volume servers.
+type Cluster struct {
+	t          testing.TB
+	weedBinary string
+	baseDir    string
+
+	masterPort     int
+	masterGrpcPort int
+	procs          []*process
+}
+
+// StartCluster boots a master and volumeServerCount empty volume servers,
+// then waits for all of them to register with the master.
+func StartCluster(t testing.TB, volumeServerCount int) *Cluster {
+	t.Helper()
+
+	weedBinary, err := findOrBuildWeedBinary()
+	if err != nil {
+		t.Fatalf("resolve weed binary: %v", err)
+	}
+
+	baseDir, err := os.MkdirTemp("", "seaweedfs_master_cold_start_it_")
+	if err != nil {
+		t.Fatalf("create temp dir: %v", err)
+	}
+	logsDir := filepath.Join(baseDir, "logs")
+	os.MkdirAll(logsDir, 0o755)
+
+	ports, err := testutil.AllocateMiniPorts(1 + volumeServerCount)
+	if err != nil {
+		t.Fatalf("allocate ports: %v", err)
+	}
+
+	c := &Cluster{
+		t:              t,
+		weedBinary:     weedBinary,
+		baseDir:        baseDir,
+		masterPort:     ports[0],
+		masterGrpcPort: ports[0] + testutil.GrpcPortOffset,
+	}
+	t.Cleanup(func() {
+		c.StopAll()
+		if t.Failed() {
+			c.DumpLogs()
+			t.Logf("cold-start logs kept at %s", baseDir)
+		} else {
+			os.RemoveAll(baseDir)
+		}
+	})
+
+	masterDir := filepath.Join(baseDir, "master")
+	os.MkdirAll(masterDir, 0o755)
+	c.startProcess("master", filepath.Join(logsDir, "master.log"),
+		"master",
+		"-ip=127.0.0.1",
+		"-port="+strconv.Itoa(c.masterPort),
+		"-port.grpc="+strconv.Itoa(c.masterGrpcPort),
+		"-mdir="+masterDir,
+		"-volumeSizeLimitMB=32",
+		"-defaultReplication=000",
+	)
+	if err := c.waitForMaster(waitTimeout); err != nil {
+		t.Fatalf("master not ready: %v", err)
+	}
+
+	for i := 0; i < volumeServerCount; i++ {
+		volDir := filepath.Join(baseDir, fmt.Sprintf("vol%d", i))
+		os.MkdirAll(volDir, 0o755)
+		name := fmt.Sprintf("volume%d", i)
+		c.startProcess(name, filepath.Join(logsDir, name+".log"),
+			"volume",
+			"-ip=127.0.0.1",
+			"-port="+strconv.Itoa(ports[1+i]),
+			"-port.grpc="+strconv.Itoa(ports[1+i]+testutil.GrpcPortOffset),
+			"-dir="+volDir,
+			"-max=10",
+			"-mserver=127.0.0.1:"+strconv.Itoa(c.masterPort),
+		)
+	}
+	if err := c.waitForVolumeServers(volumeServerCount, waitTimeout); err != nil {
+		t.Fatalf("volume servers not registered: %v", err)
+	}
+	return c
+}
+
+func (c *Cluster) startProcess(name, logFile string, args ...string) {
+	c.t.Helper()
+	f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
+	if err != nil {
+		c.t.Fatalf("create log for %s: %v", name, err)
+	}
+	defer f.Close() // the child owns its own descriptor after Start
+	cmd := exec.Command(c.weedBinary, args...)
+	cmd.Dir = c.baseDir
+	cmd.Stdout = f
+	cmd.Stderr = f
+	if err := cmd.Start(); err != nil {
+		c.t.Fatalf("start %s: %v", name, err)
+	}
+	c.procs = append(c.procs, &process{name: name, cmd: cmd, logFile: logFile})
+}
+
+// StopAll interrupts every process, volume servers first so the master does
+// not log reconnect noise.
+func (c *Cluster) StopAll() {
+	for i := len(c.procs) - 1; i >= 0; i-- {
+		p := c.procs[i]
+		if p.cmd.Process == nil {
+			continue
+		}
+		_ = p.cmd.Process.Signal(os.Interrupt)
+		done := make(chan error, 1)
+		go func() { done <- p.cmd.Wait() }()
+		select {
+		case <-time.After(10 * time.Second):
+			_ = p.cmd.Process.Kill()
+			<-done
+		case <-done:
+		}
+	}
+	c.procs = nil
+}
+
+// MasterURL returns the master HTTP endpoint.
+func (c *Cluster) MasterURL() string {
+	return fmt.Sprintf("http://127.0.0.1:%d", c.masterPort)
+}
+
+// MasterAddress returns "127.0.0.1:port" for the master HTTP port.
+func (c *Cluster) MasterAddress() string {
+	return fmt.Sprintf("127.0.0.1:%d", c.masterPort)
+}
+
+func (c *Cluster) waitForMaster(timeout time.Duration) error {
+	client := &http.Client{Timeout: 1 * time.Second}
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		resp, err := client.Get(c.MasterURL() + "/cluster/status")
+		if err == nil {
+			body, _ := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			var cs struct {
+				IsLeader bool `json:"IsLeader"`
+			}
+			if json.Unmarshal(body, &cs) == nil && cs.IsLeader {
+				return nil
+			}
+		}
+		time.Sleep(waitTick)
+	}
+	return fmt.Errorf("master did not become leader within %v", timeout)
+}
+
+// waitForVolumeServers polls /dir/status until want data nodes are registered
+// with free volume slots.
+func (c *Cluster) waitForVolumeServers(want int, timeout time.Duration) error {
+	client := &http.Client{Timeout: 2 * time.Second}
+	deadline := time.Now().Add(timeout)
+	var last string
+	for time.Now().Before(deadline) {
+		resp, err := client.Get(c.MasterURL() + "/dir/status")
+		if err == nil {
+			body, _ := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			var status struct {
+				Topology struct {
+					Max         int64 `json:"Max"`
+					DataCenters []struct {
+						Racks []struct {
+							DataNodes []struct {
+								Url string `json:"Url"`
+							} `json:"DataNodes"`
+						} `json:"Racks"`
+					} `json:"DataCenters"`
+				} `json:"Topology"`
+			}
+			if json.Unmarshal(body, &status) == nil {
+				nodes := 0
+				for _, dc := range status.Topology.DataCenters {
+					for _, rack := range dc.Racks {
+						nodes += len(rack.DataNodes)
+					}
+				}
+				if nodes >= want && status.Topology.Max > 0 {
+					return nil
+				}
+				last = fmt.Sprintf("%d/%d nodes, %d slots", nodes, want, status.Topology.Max)
+			}
+		}
+		time.Sleep(waitTick)
+	}
+	return fmt.Errorf("volume servers not registered within %v (%s)", timeout, last)
+}
+
+// DumpLogs prints the tail of every process log.
+func (c *Cluster) DumpLogs() {
+	logsDir := filepath.Join(c.baseDir, "logs")
+	entries, _ := os.ReadDir(logsDir)
+	for _, e := range entries {
+		c.t.Logf("=== %s tail ===\n%s", e.Name(), tailFile(filepath.Join(logsDir, e.Name())))
+	}
+}
+
+func tailFile(path string) string {
+	f, err := os.Open(path)
+	if err != nil {
+		return "(no log)"
+	}
+	defer f.Close()
+	scanner := bufio.NewScanner(f)
+	lines := make([]string, 0, 50)
+	for scanner.Scan() {
+		lines = append(lines, scanner.Text())
+		if len(lines) > 50 {
+			lines = lines[1:]
+		}
+	}
+	return strings.Join(lines, "\n")
+}
+
+func findOrBuildWeedBinary() (string, error) {
+	if fromEnv := os.Getenv("WEED_BINARY"); fromEnv != "" {
+		if isExecutableFile(fromEnv) {
+			return fromEnv, nil
+		}
+		return "", fmt.Errorf("WEED_BINARY not executable: %s", fromEnv)
+	}
+
+	repoRoot := ""
+	if _, file, _, ok := runtime.Caller(0); ok {
+		repoRoot = filepath.Clean(filepath.Join(filepath.Dir(file), "..", ".."))
+	}
+	if repoRoot == "" {
+		return "", fmt.Errorf("unable to detect repository root")
+	}
+
+	binDir := filepath.Join(os.TempDir(), "seaweedfs_master_cold_start_it_bin")
+	os.MkdirAll(binDir, 0o755)
+	binPath := filepath.Join(binDir, "weed")
+	if isExecutableFile(binPath) {
+		return binPath, nil
+	}
+
+	cmd := exec.Command("go", "build", "-o", binPath, ".")
+	cmd.Dir = filepath.Join(repoRoot, "weed")
+	var out bytes.Buffer
+	cmd.Stdout = &out
+	cmd.Stderr = &out
+	if err := cmd.Run(); err != nil {
+		return "", fmt.Errorf("build weed binary: %w\n%s", err, out.String())
+	}
+	return binPath, nil
+}
+
+func isExecutableFile(path string) bool {
+	info, err := os.Stat(path)
+	if err != nil || info.IsDir() {
+		return false
+	}
+	return info.Mode().Perm()&0o111 != 0
+}