diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index bdc6bc1ad..415f4fb91 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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 { diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index b2f682cc2..23ef8a5ff 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/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) + } + } +}