volume: overlap the local needle write with the replica fan-out (#10074)

* volume: overlap the local needle write with the replica fan-out

ReplicatedWrite wrote the needle to local disk and then fanned it out to
the replicas serially, so a replicated write cost local + replicas. The
needle is already fully buffered, so the two can run at once. Snapshot
the fields the fan-out reads first (the local write stamps the volume
TTL and updates needle flags, so the replica goroutine must not touch
the shared needle), then run the local write and the replica fan-out
concurrently and ack only after both. Where local disk and replica
network are separate resources the write latency drops to max(local,
replicas) instead of their sum; co-located on one disk the two writes
contend, so this helps distributed placements, not single-host tests.

* volume: return the local isUnchanged from the write goroutine

* volume: trim overlap-write comments

* volume-rs: overlap the replica fan-out with the local needle write

* volume: test the overlapped local write and replica fan-out

* volume: decode replica pairs only when replicas exist

A single-copy or incoming replicate write never uses the decoded pair map, so a malformed Pairs blob only burned an error counter and a log line. Gate the decode on replicaCount.

* volume: release the stalled replica on test failure paths

An early t.Fatal left the httptest handler blocked on releaseReplica, so the deferred Close hung instead of reporting the failure. Release idempotently ahead of the server close.

* test: mismatched If-None-Match serves the object over If-Modified-Since

The precedence case asserted the old If-Modified-Since-first ordering and
expected 304. The volume servers now evaluate If-None-Match first, so a
non-matching tag must return 200 with the body even when If-Modified-Since
would allow 304 (RFC 9110 13.1.3).

seaweed-volume/src/server/handlers.rs

@@ -3039,50 +3039,62 @@ pub async fn post_handler(
     let needle_etag = n.etag();
     let needle_has_name = n.has_name();
 
-    let write_result = write_upload_needle(&state, vid, n, fsync).await;
-
     // Replicate to remote volume servers if this volume has replicas.
     // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1.
+    let needs_replication = !is_replicate && {
+        let store = state.store.read().unwrap();
+        store
+            .find_volume(vid)
+            .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
+    };
+
+    // The fan-out runs concurrently with the local needle write, like Go's
+    // ReplicatedWrite: it reads the already-buffered request body, so the two
+    // sides share no mutable state. The ack still waits for both.
+    let replication = if needs_replication {
+        let state_clone = state.clone();
+        let path_clone = path.clone();
+        let query_clone = query.clone();
+        let headers_clone = headers.clone();
+        let body_clone = body.clone();
+        Some(tokio::spawn(async move {
+            do_replicated_request(
+                &state_clone,
+                vid.0,
+                Method::POST,
+                &path_clone,
+                &query_clone,
+                &headers_clone,
+                Some(body_clone),
+            )
+            .await
+        }))
+    } else {
+        None
+    };
+
+    let write_result = write_upload_needle(&state, vid, n, fsync).await;
+
     // No leader-known gate: when the lookup cannot run, the write must fail
-    // rather than be acknowledged on fewer copies than required.
-    if !is_replicate && write_result.is_ok() {
-        let needs_replication = {
-            let store = state.store.read().unwrap();
-            store
-                .find_volume(vid)
-                .is_some_and(|(_, v)| v.super_block().replica_placement.get_copy_count() > 1)
-        };
-        if needs_replication {
-            let state_clone = state.clone();
-            let path_clone = path.clone();
-            let query_clone = query.clone();
-            let headers_clone = headers.clone();
-            let body_clone = body.clone();
-            let replication = tokio::spawn(async move {
-                do_replicated_request(
-                    &state_clone,
-                    vid.0,
-                    Method::POST,
-                    &path_clone,
-                    &query_clone,
-                    &headers_clone,
-                    Some(body_clone),
-                )
-                .await
-            });
-            let replication_result = replication
-                .await
+    // rather than be acknowledged on fewer copies than required. The local
+    // write result takes precedence, matching Go's error ordering.
+    let replication_result = match replication {
+        Some(task) => Some(
+            task.await
                 .map_err(|e| format!("replication task failed: {}", e))
-                .flatten();
-            if let Err(e) = replication_result {
-                tracing::error!("replicated write failed: {}", e);
-                return json_error_with_query(
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    format!("replication failed: {}", e),
-                    Some(&query),
-                );
-            }
-        }
+                .flatten(),
+        ),
+        None => None,
+    };
+    if write_result.is_ok()
+        && let Some(Err(e)) = replication_result
+    {
+        tracing::error!("replicated write failed: {}", e);
+        return json_error_with_query(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            format!("replication failed: {}", e),
+            Some(&query),
+        );
     }
 
     let resp = match write_result {

weed/topology/store_replicate.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -46,8 +47,8 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 	}
 
 	// with no local volume and no replica to forward to, the write would be acknowledged without being stored
-	hasLocalVolume := s.HasVolume(volumeId)
-	if !hasLocalVolume && len(remoteLocations) == 0 {
+	vol := s.GetVolume(volumeId)
+	if vol == nil && len(remoteLocations) == 0 {
 		err = fmt.Errorf("volume %d not found on %s:%d", volumeId, s.Ip, s.Port)
 		glog.V(0).Infoln(err)
 		return
@@ -68,109 +69,141 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 		}(time.Now())
 	}
 
-	if hasLocalVolume {
-		start := time.Now()
-
-		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)
-		inFlightGauge.Inc()
-		defer inFlightGauge.Dec()
-
-		isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, true, fsync)
-		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
-		if err != nil {
-			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
-			err = fmt.Errorf("failed to write to local disk: %w", err)
-			glog.V(0).Infoln(err)
-			return
+	// Snapshot the fields the fan-out reads before the concurrent local write
+	// mutates the needle (it stamps the volume TTL and flags), so the two cannot
+	// race; the fan-out then touches only these locals plus read-only n.Data.
+	replicaTtl := n.Ttl
+	if vol != nil && replicaTtl == needle.EMPTY_TTL && vol.Ttl != needle.EMPTY_TTL {
+		replicaTtl = vol.Ttl
+	}
+	replicaLastModified := n.LastModified
+	replicaIsChunked := n.IsChunkedManifest()
+	replicaIsCompressed := n.IsCompressed()
+	replicaName := string(n.Name)
+	replicaMime := string(n.Mime)
+	replicaData := n.Data
+	replicaPairMap := make(map[string]string)
+	if replicaCount > 0 && n.HasPairs() {
+		tmpMap := make(map[string]string)
+		if pe := json.Unmarshal(n.Pairs, &tmpMap); pe != nil {
+			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
+			glog.V(0).Infoln("Unmarshal pairs error:", pe)
+		}
+		for k, v := range tmpMap {
+			replicaPairMap[needle.PairNamePrefix+k] = v
 		}
 	}
 
 	// Observe replication targets histogram for all operations (including zero)
 	stats.VolumeServerReplicationTargets.Observe(float64(replicaCount))
 
-	if replicaCount > 0 { //send to other replica locations
-		start := time.Now()
+	// Run the local disk write and the replica fan-out concurrently against the
+	// buffered needle (latency max(local, replicas), not the sum); ack after both.
+	var wg sync.WaitGroup
+	var localErr, remoteErr error
+	var localIsUnchanged bool
 
-		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas)
-		inFlightGauge.Inc()
-		defer inFlightGauge.Dec()
+	if vol != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			start := time.Now()
+			inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)
+			inFlightGauge.Inc()
+			defer inFlightGauge.Dec()
 
-		err = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error {
-			u := url.URL{
-				Scheme: "http",
-				Host:   location.Url,
-				Path:   r.URL.Path,
+			localIsUnchanged, localErr = s.WriteVolumeNeedle(volumeId, n, true, fsync)
+			stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
+			if localErr != nil {
+				stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
+				localErr = fmt.Errorf("failed to write to local disk: %w", localErr)
+				glog.V(0).Infoln(localErr)
 			}
-			q := url.Values{
-				"type": {"replicate"},
-				"ttl":  {n.Ttl.String()},
-			}
-			if fsync {
-				q.Set("fsync", "true")
-			}
-			if n.LastModified > 0 {
-				q.Set("ts", strconv.FormatUint(n.LastModified, 10))
-			}
-			if n.IsChunkedManifest() {
-				q.Set("cm", "true")
-			}
-			u.RawQuery = q.Encode()
-
-			pairMap := make(map[string]string)
-			if n.HasPairs() {
-				tmpMap := make(map[string]string)
-				err := json.Unmarshal(n.Pairs, &tmpMap)
-				if err != nil {
-					stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
-					glog.V(0).Infoln("Unmarshal pairs error:", err)
-				}
-				for k, v := range tmpMap {
-					pairMap[needle.PairNamePrefix+k] = v
-				}
-			}
-			bytesBuffer := buffer_pool.SyncPoolGetBuffer()
-			defer buffer_pool.SyncPoolPutBuffer(bytesBuffer)
-
-			// volume server do not know about encryption
-			// TODO optimize here to compress data only once
-			uploadOption := &operation.UploadOption{
-				UploadUrl:         u.String(),
-				Filename:          string(n.Name),
-				Cipher:            false,
-				IsInputCompressed: n.IsCompressed(),
-				IsReplication:     true,
-				MimeType:          string(n.Mime),
-				PairMap:           pairMap,
-				Jwt:               jwt,
-				Md5:               contentMd5,
-				BytesBuffer:       bytesBuffer,
-				MaxAttempts:       1, // fail fast on a dead replica; the client write retries
-			}
-
-			uploader, err := operation.NewUploader()
-			if err != nil {
-				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
-				return err
-			}
-			_, err = uploader.UploadData(ctx, n.Data, uploadOption)
-			if err != nil {
-				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
-			}
-			return err
-		})
-		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
-		if err != nil {
-			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
-			stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc()
-			reason := classifyReplicationError(err)
-			stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc()
-			err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
-			glog.V(0).Infoln(err)
-			return false, err
-		}
-		stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc()
+		}()
 	}
-	return
+
+	if replicaCount > 0 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			start := time.Now()
+			inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToReplicas)
+			inFlightGauge.Inc()
+			defer inFlightGauge.Dec()
+
+			remoteErr = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error {
+				u := url.URL{
+					Scheme: "http",
+					Host:   location.Url,
+					Path:   r.URL.Path,
+				}
+				q := url.Values{
+					"type": {"replicate"},
+					"ttl":  {replicaTtl.String()},
+				}
+				if fsync {
+					q.Set("fsync", "true")
+				}
+				if replicaLastModified > 0 {
+					q.Set("ts", strconv.FormatUint(replicaLastModified, 10))
+				}
+				if replicaIsChunked {
+					q.Set("cm", "true")
+				}
+				u.RawQuery = q.Encode()
+
+				bytesBuffer := buffer_pool.SyncPoolGetBuffer()
+				defer buffer_pool.SyncPoolPutBuffer(bytesBuffer)
+
+				// volume server do not know about encryption
+				uploadOption := &operation.UploadOption{
+					UploadUrl:         u.String(),
+					Filename:          replicaName,
+					Cipher:            false,
+					IsInputCompressed: replicaIsCompressed,
+					IsReplication:     true,
+					MimeType:          replicaMime,
+					PairMap:           replicaPairMap,
+					Jwt:               jwt,
+					Md5:               contentMd5,
+					BytesBuffer:       bytesBuffer,
+					MaxAttempts:       1, // fail fast on a dead replica; the client write retries
+				}
+
+				uploader, uploaderErr := operation.NewUploader()
+				if uploaderErr != nil {
+					glog.Errorf("replication-UploadData, err:%v, url:%s", uploaderErr, u.String())
+					return uploaderErr
+				}
+				_, uploadErr := uploader.UploadData(ctx, replicaData, uploadOption)
+				if uploadErr != nil {
+					glog.Errorf("replication-UploadData, err:%v, url:%s", uploadErr, u.String())
+				}
+				return uploadErr
+			})
+			stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
+			if remoteErr != nil {
+				stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
+				stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationFailure).Inc()
+				reason := classifyReplicationError(remoteErr)
+				stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, reason).Inc()
+				remoteErr = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, remoteErr)
+				glog.V(0).Infoln(remoteErr)
+			} else {
+				stats.VolumeServerReplicationCounter.WithLabelValues(stats.ReplicationOpWrite, stats.ReplicationSuccess).Inc()
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	if localErr != nil {
+		return false, localErr
+	}
+	if remoteErr != nil {
+		return false, remoteErr
+	}
+	return localIsUnchanged, nil
 }
 
 // ReplicatedDelete deletes a needle from the local volume and sends delete

weed/topology/store_replicate_test.go

@@ -16,8 +16,11 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -230,6 +233,176 @@ func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) {
 	}
 }
 
+// newReplicateTestStore builds a single-disk store in dir, draining the notify
+// channels so volume mounts never block on a heartbeat reader.
+func newReplicateTestStore(t *testing.T, dir string) *storage.Store {
+	t.Helper()
+	store := storage.NewStore(nil, "127.0.0.1", 8080, 18080, "http://127.0.0.1:8080", "store-id",
+		[]string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "",
+		storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3,
+		stats.DefaultDiskIOProbeConfig())
+	done := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-store.NewVolumesChan:
+			case <-store.DeletedVolumesChan:
+			case <-store.NewEcShardsChan:
+			case <-store.DeletedEcShardsChan:
+			case <-store.StateUpdateChan:
+			case <-done:
+				return
+			}
+		}
+	}()
+	t.Cleanup(func() {
+		close(done)
+		store.Close()
+	})
+	return store
+}
+
+// TestReplicatedWriteOverlapsLocalAndReplicaWrites runs the write path with a
+// real local volume and a replica that stalls mid-request: the fan-out starts
+// while the local write is still free to run, the local write lands before the
+// replica unblocks, and the acknowledged write waits for both legs.
+func TestReplicatedWriteOverlapsLocalAndReplicaWrites(t *testing.T) {
+	replicaStarted := make(chan url.Values, 1)
+	releaseReplica := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseReplica) }) }
+	replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		replicaStarted <- r.URL.Query()
+		<-releaseReplica
+		w.WriteHeader(http.StatusCreated)
+		if _, err := w.Write([]byte(`{"size":21}`)); err != nil {
+			t.Errorf("replica write response: %v", err)
+		}
+	}))
+	defer replica.Close()
+	defer release()
+	replicaHost := strings.TrimPrefix(replica.URL, "http://")
+
+	masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{
+		locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}},
+	})
+
+	store := newReplicateTestStore(t, t.TempDir())
+	volumeId := needle.VolumeId(7)
+	operation.InvalidateVolumeIdLocationCache(volumeId.String())
+	if err := store.AddVolume(volumeId, "", storage.NeedleMapInMemory, "001", "15d", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	data := []byte("overlapped write data")
+	n := &needle.Needle{
+		Id:       1,
+		Data:     data,
+		Checksum: needle.NewCRC(data),
+		Ttl:      needle.EMPTY_TTL,
+	}
+	r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/7,01637037d6", nil)
+	if err := r.ParseForm(); err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, volumeId, n, r, "")
+		done <- err
+	}()
+
+	var q url.Values
+	select {
+	case q = <-replicaStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("replica never received the fan-out request")
+	}
+	if got := q.Get("type"); got != "replicate" {
+		t.Errorf("replica type = %q, want replicate", got)
+	}
+	if got := q.Get("ttl"); got != "15d" {
+		t.Errorf("replica ttl = %q, want the inherited volume ttl 15d", got)
+	}
+
+	select {
+	case err := <-done:
+		t.Fatalf("write returned before the replica responded: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		readBack := &needle.Needle{Id: n.Id}
+		if _, err := store.ReadVolumeNeedle(volumeId, readBack, nil, nil); err == nil && string(readBack.Data) == string(data) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("local write did not land while the replica was still blocked")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	release()
+	if err := <-done; err != nil {
+		t.Fatalf("ReplicatedWrite: %v", err)
+	}
+}
+
+// TestReplicatedWriteFailsWhenEitherLegFails checks that a failed local write
+// or a failed replica fan-out fails the whole request even when the other leg
+// succeeded.
+func TestReplicatedWriteFailsWhenEitherLegFails(t *testing.T) {
+	for _, tc := range []struct {
+		name          string
+		volumeId      needle.VolumeId
+		readonlyLocal bool
+		replicaStatus int
+	}{
+		{name: "local write fails", volumeId: 8, readonlyLocal: true, replicaStatus: http.StatusCreated},
+		{name: "replica write fails", volumeId: 9, replicaStatus: http.StatusInternalServerError},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(tc.replicaStatus)
+			}))
+			defer replica.Close()
+			replicaHost := strings.TrimPrefix(replica.URL, "http://")
+
+			masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{
+				locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}, {Url: replicaHost}},
+			})
+
+			store := newReplicateTestStore(t, t.TempDir())
+			operation.InvalidateVolumeIdLocationCache(tc.volumeId.String())
+			if err := store.AddVolume(tc.volumeId, "", storage.NeedleMapInMemory, "001", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil {
+				t.Fatal(err)
+			}
+			if tc.readonlyLocal {
+				if err := store.MarkVolumeReadonly(tc.volumeId, false, false); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			data := []byte("write data")
+			n := &needle.Needle{
+				Id:       1,
+				Data:     data,
+				Checksum: needle.NewCRC(data),
+				Ttl:      needle.EMPTY_TTL,
+			}
+			r := httptest.NewRequest(http.MethodPost, fmt.Sprintf("http://127.0.0.1:8080/%d,01637037d6", tc.volumeId), nil)
+			if err := r.ParseForm(); err != nil {
+				t.Fatal(err)
+			}
+
+			if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, tc.volumeId, n, r, ""); err == nil {
+				t.Fatal("ReplicatedWrite succeeded despite a failed leg")
+			}
+		})
+	}
+}
+
 // TestReplicatedWriteRejectsWriteWithNoTarget verifies that a write this server
 // cannot store and has no replica to forward to fails instead of being
 // acknowledged with nothing written.