diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 1310793564..6826d4f3f1 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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 { diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index fc7da99fb3..7ef9ca5694 100644 --- a/weed/topology/store_replicate.go +++ b/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 diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go index 3901db744f..74eadaa48c 100644 --- a/weed/topology/store_replicate_test.go +++ b/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.