fix(volume): return an error instead of 201 when a write lands on no volume (#11397)

* fix(volume): return an error instead of 201 when a write lands on no volume

ReplicatedWrite only writes locally when this server holds the volume.
For a volume id no server holds, the master lookup returns no locations,
so the write went nowhere and the upload still got 201 Created. The same
happened for a type=replicate write to a server without the volume, so
the primary, or the S3 chunk fan-out, counted a replica that was never
written.

A server without the volume still forwards the write to the replicas the
master lists. When there is nothing to forward to, fail with "volume N
not found on host:port". PostHandler returns that as 500, the status the
Rust volume server already returns here, and uploaders re-assign on 5xx.

Fixes #6609

* volume: reuse Store.HasVolume, drop issue ref from test comment

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/topology/store_replicate.go

@@ -45,6 +45,14 @@ 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 {
+		err = fmt.Errorf("volume %d not found on %s:%d", volumeId, s.Ip, s.Port)
+		glog.V(0).Infoln(err)
+		return
+	}
+
 	// read fsync value
 	fsync := false
 	if r.FormValue("fsync") == "true" {
@@ -60,7 +68,7 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 		}(time.Now())
 	}
 
-	if s.GetVolume(volumeId) != nil {
+	if hasLocalVolume {
 		start := time.Now()
 
 		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)

weed/topology/store_replicate_test.go

@@ -80,10 +80,14 @@ func (m *mockMasterServer) LookupVolume(ctx context.Context, req *master_pb.Look
 	m.calls++
 	var vls []*master_pb.LookupVolumeResponse_VolumeIdLocation
 	for _, vid := range req.VolumeOrFileIds {
-		vls = append(vls, &master_pb.LookupVolumeResponse_VolumeIdLocation{
+		vl := &master_pb.LookupVolumeResponse_VolumeIdLocation{
 			VolumeOrFileId: vid,
 			Locations:      m.locations,
-		})
+		}
+		if len(m.locations) == 0 {
+			vl.Error = fmt.Sprintf("volume id %s not found", vid)
+		}
+		vls = append(vls, vl)
 	}
 	return &master_pb.LookupVolumeResponse{VolumeIdLocations: vls}, nil
 }
@@ -225,3 +229,46 @@ func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) {
 		})
 	}
 }
+
+// TestReplicatedWriteRejectsWriteWithNoTarget verifies that a write this server
+// cannot store and has no replica to forward to fails instead of being
+// acknowledged with nothing written.
+func TestReplicatedWriteRejectsWriteWithNoTarget(t *testing.T) {
+	master := &mockMasterServer{}
+	masterFn, dialOption := startMockMasterServer(t, master)
+	store := &storage.Store{Ip: "127.0.0.1", Port: 8080}
+	volumeId := needle.VolumeId(31234)
+
+	for _, tc := range []struct {
+		name      string
+		query     string
+		locations []*master_pb.Location
+	}{
+		{name: "volume unknown to the master"},
+		{name: "master lists only this server", locations: []*master_pb.Location{{Url: "127.0.0.1:8080"}}},
+		{name: "replica write for a volume not on this server", query: "?type=replicate"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			master.mu.Lock()
+			master.locations = tc.locations
+			master.mu.Unlock()
+
+			r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/31234,01637037d6"+tc.query, nil)
+			if err := r.ParseForm(); err != nil {
+				t.Fatal(err)
+			}
+			n := &needle.Needle{
+				Id:   1,
+				Data: []byte("test data"),
+				Ttl:  needle.EMPTY_TTL,
+			}
+			_, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, volumeId, n, r, "")
+			if err == nil {
+				t.Fatal("ReplicatedWrite acknowledged a write that no volume stored")
+			}
+			if !strings.Contains(err.Error(), "not found") {
+				t.Fatalf("unexpected error: %v", err)
+			}
+		})
+	}
+}