ensure monotonic n.AppendAtNs in each place (#3880)

https://github.com/seaweedfs/seaweedfs/issues/3852

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

2 files changed, 19 insertions, 12 deletions

diff --git a/weed/storage/needle/needle.go b/weed/storage/needle/needle.go
index 80c0b7b51..9aa684b3c 100644
--- a/weed/storage/needle/needle.go
+++ b/weed/storage/needle/needle.go
@@ -142,6 +142,14 @@ func (n *Needle) ParsePath(fid string) (err error) {
 	return err
 }
 
+func GetAppendAtNs(volumeLastAppendAtNs uint64) uint64 {
+	return max(uint64(time.Now().UnixNano()), volumeLastAppendAtNs+1)
+}
+
+func (n *Needle) UpdateAppendAtNs(volumeLastAppendAtNs uint64) {
+	n.AppendAtNs = max(uint64(time.Now().UnixNano()), volumeLastAppendAtNs+1)
+}
+
 func ParseNeedleIdCookie(key_hash_string string) (NeedleId, Cookie, error) {
 	if len(key_hash_string) <= CookieSize*2 {
 		return NeedleIdEmpty, 0, fmt.Errorf("KeyHash is too short.")
@@ -164,3 +172,10 @@ func ParseNeedleIdCookie(key_hash_string string) (NeedleId, Cookie, error) {
 func (n *Needle) LastModifiedString() string {
 	return time.Unix(int64(n.LastModified), 0).Format("2006-01-02T15:04:05")
 }
+
+func max(x, y uint64) uint64 {
+	if x <= y {
+		return y
+	}
+	return x
+}
diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go
index 954b06159..e2537521f 100644
--- a/weed/storage/volume_write.go
+++ b/weed/storage/volume_write.go
@@ -4,13 +4,11 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"os"
-	"time"
-
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"os"
 )
 
 var ErrorNotFound = errors.New("not found")
@@ -157,7 +155,7 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool) (offset uint
 	}
 
 	// append to dat file
-	n.AppendAtNs = max(uint64(time.Now().UnixNano()), v.lastAppendAtNs+1)
+	n.UpdateAppendAtNs(v.lastAppendAtNs)
 	offset, size, _, err = n.Append(v.DataBackend, v.Version())
 	v.checkReadWriteError(err)
 	if err != nil {
@@ -218,6 +216,7 @@ func (v *Volume) doDeleteRequest(n *needle.Needle) (Size, error) {
 		size := nv.Size
 		if !v.hasRemoteFile {
 			n.Data = nil
+			n.UpdateAppendAtNs(v.lastAppendAtNs)
 			n.AppendAtNs = uint64(time.Now().UnixNano())
 			offset, _, _, err = n.Append(v.DataBackend, v.Version())
 			v.checkReadWriteError(err)
@@ -318,7 +317,7 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 		return fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.nm.ContentSize())
 	}
 
-	appendAtNs := uint64(time.Now().UnixNano())
+	appendAtNs := needle.GetAppendAtNs(v.lastAppendAtNs)
 	offset, err := needle.WriteNeedleBlob(v.DataBackend, needleBlob, size, appendAtNs, v.Version())
 
 	v.checkReadWriteError(err)
@@ -334,10 +333,3 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 
 	return err
 }
-
-func max(x, y uint64) uint64 {
-	if x <= y {
-		return y
-	}
-	return x
-}