persist readonly state to volume info (#5977)

weed/pb/volume_server.proto

@@ -480,6 +480,7 @@ message VolumeInfo {
     uint32 BytesOffset = 4;
     int64 dat_file_size = 5; // used for EC encoded volumes to store the original file size
     uint64 DestroyTime = 6; // used to record the destruction time of ec volume
+    bool read_only = 7;
 }
 
 // tiered storage

weed/storage/store.go

@@ -477,6 +477,7 @@ func (s *Store) MarkVolumeReadonly(i needle.VolumeId) error {
 	}
 	v.noWriteLock.Lock()
 	v.noWriteOrDelete = true
+	v.PersistReadOnly(true)
 	v.noWriteLock.Unlock()
 	return nil
 }
@@ -488,6 +489,7 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	}
 	v.noWriteLock.Lock()
 	v.noWriteOrDelete = false
+	v.PersistReadOnly(false)
 	v.noWriteLock.Unlock()
 	return nil
 }

weed/storage/volume.go

@@ -48,8 +48,9 @@ type Volume struct {
 	isCompacting       bool
 	isCommitCompacting bool
 
-	volumeInfo *volume_server_pb.VolumeInfo
-	location   *DiskLocation
+	volumeInfoRWLock sync.RWMutex
+	volumeInfo       *volume_server_pb.VolumeInfo
+	location         *DiskLocation
 
 	lastIoError error
 }
@@ -358,3 +359,10 @@ func (v *Volume) IsReadOnly() bool {
 	defer v.noWriteLock.RUnlock()
 	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
 }
+
+func (v *Volume) PersistReadOnly(readOnly bool) {
+	v.volumeInfoRWLock.RLock()
+	defer v.volumeInfoRWLock.RUnlock()
+	v.volumeInfo.ReadOnly = readOnly
+	v.SaveVolumeInfo()
+}

weed/storage/volume_loading.go

@@ -43,6 +43,11 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 
 	hasVolumeInfoFile := v.maybeLoadVolumeInfo()
 
+	if v.volumeInfo.ReadOnly && !v.HasRemoteFile() {
+		// this covers the case where the volume is marked as read-only and has no remote file
+		v.noWriteOrDelete = true
+	}
+
 	if v.HasRemoteFile() {
 		v.noWriteCanDelete = true
 		v.noWriteOrDelete = false