volume: reject needle blob writes to read-only volumes (#10435)

* volume: reject needle blob writes to read-only volumes

WriteNeedleBlob appends the blob to .dat and only then calls nm.Put. On a
read-only volume the needle map is a SortedFileNeedleMap whose Put always
fails, so the append is never indexed and never rolled back.

Nothing upstream stops this: volume.check.disk picks its targets from the
master's cached topology, which goes stale the moment a volume server marks
a replica read-only itself — a failed data integrity check at load, or an
EIO quarantine. Each sync attempt then grows the .dat of a replica that is
supposed to be frozen by one unindexed needle, and reports it as "invalid
argument", the bare os.ErrInvalid the needle map returns.

Check IsReadOnly before touching .dat, same as the upload path does.

* volume: say which needle and volume failed to index

An index write that fails surfaced as a bare errno with no volume, no needle
and no file — "invalid argument" for a read-only needle map, or a plain
ENOSPC when .idx lives on its own filesystem via -dir.idx. Both were logged
at V(4), so by default the operator saw only the errno the client got back.

seaweed-volume/src/storage/volume.rs

@@ -2713,6 +2713,11 @@ impl Volume {
         needle_blob: &[u8],
         size: Size,
     ) -> Result<(), VolumeError> {
+        // nm.put on a read-only volume fails only after the blob is appended to .dat.
+        if self.is_read_only() {
+            return Err(VolumeError::ReadOnly);
+        }
+
         // Dedup check: if the same needle already exists with matching content, skip the write.
         // Matches Go's WriteNeedleBlob which reads existing needle and compares cookie+checksum+data.
         if let Some(nm) = &self.nm {

weed/storage/needle_map_sorted_file.go

@@ -78,7 +78,7 @@ func (m *SortedFileNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue
 }
 
 func (m *SortedFileNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
-	return os.ErrInvalid
+	return fmt.Errorf("needle map %s.sdx is read only: %w", m.baseFileName, os.ErrInvalid)
 }
 
 func (m *SortedFileNeedleMap) Delete(key NeedleId, offset Offset) error {

weed/storage/volume_write.go

@@ -228,7 +228,8 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool) (offset uint
 	// add to needle map
 	if !ok || uint64(nv.Offset.ToActualOffset()) < offset {
 		if err = v.nm.Put(n.Id, ToOffset(int64(offset)), n.Size); err != nil {
-			glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err)
+			err = fmt.Errorf("index needle %d of volume %d at offset %d: %w", n.Id, v.Id, offset, err)
+			glog.V(0).Info(err)
 		}
 	}
 	if v.lastModifiedTsSeconds < n.LastModified {
@@ -372,6 +373,11 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
+	// nm.Put on a read-only volume fails only after the blob is appended to .dat.
+	if v.IsReadOnly() {
+		return fmt.Errorf("volume %d is read only", v.Id)
+	}
+
 	if MaxPossibleVolumeSize < v.nm.ContentSize()+uint64(len(needleBlob)) {
 		return fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.nm.ContentSize())
 	}
@@ -400,7 +406,8 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 
 	// add to needle map
 	if err = v.nm.Put(needleId, ToOffset(int64(offset)), size); err != nil {
-		glog.V(4).Infof("failed to put in needle map %d: %v", needleId, err)
+		err = fmt.Errorf("index needle %d of volume %d at offset %d: %w", needleId, v.Id, offset, err)
+		glog.V(0).Info(err)
 	}
 
 	return err

weed/storage/volume_write_test.go

@@ -167,3 +167,47 @@ func TestDestroyNonemptyVolumeWithoutOnlyEmpty(t *testing.T) {
 	}
 	assertFileExist(t, false, path)
 }
+
+// Pre-fix: the blob was appended to .dat, then rejected by SortedFileNeedleMap.Put.
+func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) {
+	dir := t.TempDir()
+
+	v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	offset, size, _, err := v.writeNeedle2(newRandomNeedle(1), true, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	blob, err := v.ReadNeedleBlob(int64(offset), size)
+	if err != nil {
+		t.Fatalf("read needle blob: %v", err)
+	}
+	v.PersistReadOnly(true)
+	v.Close()
+
+	v, err = NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume reload: %v", err)
+	}
+	defer v.Close()
+	if _, ok := v.nm.(*SortedFileNeedleMap); !ok {
+		t.Fatalf("reloaded read-only volume should use SortedFileNeedleMap, got %T", v.nm)
+	}
+
+	datSizeBefore, _, _ := v.DataBackend.GetStat()
+
+	err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, size)
+	if err == nil {
+		t.Fatalf("expected WriteNeedleBlob to be rejected on a read-only volume")
+	}
+	if errors.Is(err, os.ErrInvalid) {
+		t.Errorf("WriteNeedleBlob should fail with a read-only error, not the needle map's os.ErrInvalid: %v", err)
+	}
+
+	datSizeAfter, _, _ := v.DataBackend.GetStat()
+	if datSizeAfter != datSizeBefore {
+		t.Errorf("read-only volume .dat grew from %d to %d, leaving an unindexed needle", datSizeBefore, datSizeAfter)
+	}
+}