filer.remote.sync: stamp entries with IF_CHUNKS_EQUAL so a stale write-back cannot delete live chunks

updateLocalEntry records the RemoteEntry stamp after an upload by writing the
event's entry back with UpdateEntry. The filer deletes every stored chunk
absent from an updated entry, so when the file was rewritten while its upload
was in flight (or the event is a replay), the stale snapshot deletes the
rewrite's chunks: the entry then points at the new fid with no needle behind
it, and the rewrite's own upload fails and is skipped as superseded.

The stamp write now carries WriteCondition IF_CHUNKS_EQUAL over the event's
chunk fids, evaluated by the filer under the path lock. A refused stamp means
the filer moved past this event; the superseding event follows in the log and
stamps the current entry, so the refusal is logged and skipped like a
superseded upload.

Reproduction: weed server -filer plus a weed server -s3 remote, remote.mount,
filer.remote.sync; hold the remote (docker pause) so one upload stays in
flight, rewrite the file through the filer, unpause. Before: the entry's chunk
is 404 on every volume server. After: the stale stamp is refused, the rewrite's
chunk stays live and reads back after a vacuum.

weed/command/filer_remote_sync_dir.go

@@ -20,6 +20,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -417,16 +419,57 @@ func shouldSendToRemote(entry *filer_pb.Entry) bool {
 	return false
 }
 
+// updateLocalEntry stamps the entry an event described with its RemoteEntry.
+// The write carries IF_CHUNKS_EQUAL over the event's chunk fids: the filer
+// deletes every stored chunk absent from an updated entry, so a snapshot
+// older than the live entry (the file was rewritten while its upload was in
+// flight, or the event is a replay) would delete the live chunks. A failed
+// precondition means the filer moved past this event; the event that
+// superseded it follows in the log and stamps the current entry, so the stale
+// stamp is skipped the same way a superseded upload is.
 func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error {
 	remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano()
 	entry.RemoteEntry = remoteEntry
-	return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		_, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
 			Directory: dir,
 			Entry:     entry,
+			Condition: ifChunksEqual(entry),
 		})
 		return err
 	})
+	if isFailedPrecondition(err) {
+		glog.Errorf("skipping stale stamp of %s: %v", util.NewFullPath(dir, entry.Name), err)
+		return nil
+	}
+	return err
+}
+
+// ifChunksEqual builds the precondition that the stored entry still holds
+// exactly the chunks the event described. An entry with inline content and no
+// chunks yields an empty fid list, which the filer satisfies only while the
+// stored entry has no chunks either.
+func ifChunksEqual(entry *filer_pb.Entry) *filer_pb.WriteCondition {
+	chunks := entry.GetChunks()
+	fids := make([]string, 0, len(chunks))
+	for _, chunk := range chunks {
+		fids = append(fids, chunk.GetFileIdString())
+	}
+	return &filer_pb.WriteCondition{
+		Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_CHUNKS_EQUAL, Fids: fids}},
+	}
+}
+
+// isFailedPrecondition reports a write condition the filer refused, through
+// any wrapping WithFilerClient added.
+func isFailedPrecondition(err error) bool {
+	if err == nil {
+		return false
+	}
+	if st, ok := status.FromError(err); ok && st.Code() == codes.FailedPrecondition {
+		return true
+	}
+	return strings.Contains(err.Error(), "precondition failed")
 }
 
 func isMultipartUploadFile(dir string, name string) bool {

weed/command/filer_remote_sync_dir_test.go

@@ -16,6 +16,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -808,3 +810,47 @@ func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) {
 		t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite))
 	}
 }
+
+func TestIfChunksEqualCarriesTheEventChunkFids(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Name: "f",
+		Chunks: []*filer_pb.FileChunk{
+			{FileId: "3,01a"},
+			{Fid: &filer_pb.FileId{VolumeId: 4, FileKey: 0x2b, Cookie: 0x0c}},
+		},
+	}
+	cond := ifChunksEqual(entry)
+	if len(cond.Clauses) != 1 || cond.Clauses[0].Kind != filer_pb.WriteCondition_IF_CHUNKS_EQUAL {
+		t.Fatalf("condition = %v, want one IF_CHUNKS_EQUAL clause", cond)
+	}
+	got := cond.Clauses[0].Fids
+	want := []string{"3,01a", entry.Chunks[1].GetFileIdString()}
+	if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
+		t.Fatalf("fids = %v, want %v", got, want)
+	}
+
+	inline := &filer_pb.Entry{Name: "small", Content: []byte("bytes")}
+	if fids := ifChunksEqual(inline).Clauses[0].Fids; len(fids) != 0 {
+		t.Fatalf("inline entry fids = %v, want none", fids)
+	}
+}
+
+func TestIsFailedPrecondition(t *testing.T) {
+	refused := status.Error(codes.FailedPrecondition, "precondition failed: /buckets/b/f")
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"status", refused, true},
+		{"wrapped status", fmt.Errorf("update entry: %w", refused), true},
+		{"message only", errors.New("rpc error: code = FailedPrecondition desc = precondition failed: /f"), true},
+		{"other", status.Error(codes.Unavailable, "filer down"), false},
+	}
+	for _, c := range cases {
+		if got := isFailedPrecondition(c.err); got != c.want {
+			t.Errorf("%s: isFailedPrecondition = %v, want %v", c.name, got, c.want)
+		}
+	}
+}