ensure only compatible volume versions are writable

3 files changed, 8 insertions, 3 deletions

diff --git a/weed-fs/src/pkg/storage/volume_info.go b/weed-fs/src/pkg/storage/volume_info.go
index 060277e0a..e4c5f6ec4 100644
--- a/weed-fs/src/pkg/storage/volume_info.go
+++ b/weed-fs/src/pkg/storage/volume_info.go
@@ -6,6 +6,7 @@ type VolumeInfo struct {
 	Id               VolumeId
 	Size             uint64
 	RepType          ReplicationType
+	Version          Version
 	FileCount        int
 	DeleteCount      int
 	DeletedByteCount uint64
diff --git a/weed-fs/src/pkg/topology/topology_event_handling.go b/weed-fs/src/pkg/topology/topology_event_handling.go
index 08c08fcde..b33b4d768 100644
--- a/weed-fs/src/pkg/topology/topology_event_handling.go
+++ b/weed-fs/src/pkg/topology/topology_event_handling.go
@@ -59,8 +59,8 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) {
 }
 func (t *Topology) RegisterRecoveredDataNode(dn *DataNode) {
 	for _, v := range dn.volumes {
-		if uint64(v.Size) < t.volumeSizeLimit {
-			vl := t.GetVolumeLayout(v.RepType)
+        vl := t.GetVolumeLayout(v.RepType)
+		if vl.isWritable(&v) {
 			vl.SetVolumeAvailable(dn, v.Id)
 		}
 	}
diff --git a/weed-fs/src/pkg/topology/volume_layout.go b/weed-fs/src/pkg/topology/volume_layout.go
index 535effde6..80eff5061 100644
--- a/weed-fs/src/pkg/topology/volume_layout.go
+++ b/weed-fs/src/pkg/topology/volume_layout.go
@@ -31,13 +31,17 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 	}
 	if vl.vid2location[v.Id].Add(dn) {
 		if len(vl.vid2location[v.Id].list) == v.RepType.GetCopyCount() {
-			if uint64(v.Size) < vl.volumeSizeLimit {
+			if vl.isWritable(v) {
 				vl.writables = append(vl.writables, v.Id)
 			}
 		}
 	}
 }
 
+func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool{
+    return uint64(v.Size) < vl.volumeSizeLimit && v.Version == storage.CurrentVersion
+}
+
 func (vl *VolumeLayout) Lookup(vid storage.VolumeId) *[]*DataNode {
 	return &vl.vid2location[vid].list
 }