s3api: don't delete chunks when CreateEntry outcome is ambiguous (#11376)

* s3api: map ambiguous filer transport errors to retryable 503

Canceled, DeadlineExceeded and Unavailable can be returned after the
filer applied the write, so the outcome is ambiguous. Reporting them as
a 4xx tells the client not to retry; report ServiceUnavailable instead.

* s3api: verify entry existence before deleting orphaned chunks

A failed CreateEntry can still have landed on the filer when the error
is a transport failure, and entryCreated=false would tombstone chunks a
live entry references, leaving a dangling pointer that survives only
because reads pass readDeleted=true until vacuum reclaims the needle.

Before deleting, look the entry up: if it is stored with the same
chunks, the write succeeded; if the lookup cannot be answered, keep the
chunks for vacuum to reclaim; only a confirmed absence still cleans up.

* s3api: regression tests for ambiguous CreateEntry outcomes

Covers the three post-create-failure cases in putToFiler: the entry
landed despite the error (treat as success, keep chunks), the entry is
confirmed absent (delete orphans), and the outcome is unverifiable
(keep chunks, return error).

* volume: count reads served from deleted needles

A readDeleted read succeeding on a tombstoned needle is the signal that
metadata still points at deleted data. Count it under a
readDeletedNeedle handler label in both the Go and Rust volume servers
so the condition is visible before vacuum turns it into a 404.

* s3api: never delete chunks on an ambiguous create error

Review feedback on the first fix showed verification could still go
wrong in both directions: a stale or lagged lookup could report
not-found for a committed entry, a prefix object stores its chunks on a
directory entry, and filer-side manifestization rewrites the top-level
chunk ids the comparison relied on.

Rework the rule so the outcome classes are asymmetric:

- A transport-level error (anything filerErrorToS3Error maps to a
  retryable 503) is ambiguous and never deletes chunks; the lookup can
  only upgrade the write to success.
- Any other error is a definitive filer refusal and still cleans up.

confirmCreateLanded asks the write owner first, resolves the stored
entry through chunk manifests, requires an exact match of the uploaded
file ids, and on success runs the finalize callback the failed create
skipped (under the object write lock, with the same rmObject undo the
create path uses). Zero-chunk writes stay ambiguous since they cannot
be told apart by chunks.

* s3api: cover definitive refusals and stale entries in put tests

The confirmed-failure case now uses a definitive refusal so it still
exercises orphan cleanup, and a new case keeps chunks when the stored
entry belongs to an older object rather than this PUT.

* volume: count deleted-needle reads once per request

Streamed Go reads ran the deleted check in readNeedle and again in
readNeedleDataInto, and non-streamed Rust reads in stream_info and the
full-read fallback, double-counting one request. Count at the single
entry probe each implementation takes per GET: readNeedle in Go,
read_needle_stream_info in Rust.

* s3api: run recovered-write rollback under the object lock

Two follow-ups from review: ResolveChunkManifest returns traversed
manifest blobs in its manifestChunks output, so requiring it empty
rejected every manifestized landing; and the rmObject undo ran after
the object write lock was released, so a concurrent newer write could
be deleted between finalize failure and rollback. Compare only the
resolved data chunks and keep the undo inside the lock.

* s3api: verify, finalize and roll back recovered creates in one lock

A lookup done before the object write lock let a concurrent PUT replace
the entry between the chunk comparison and the finalize/rollback
section, so a failed afterCreate could rmObject a newer write. Run the
owner lookup, manifest resolution, chunk comparison, afterCreate and
the conditional undo inside a single withObjectWriteLock section.

seaweed-volume/src/metrics.rs

@@ -349,6 +349,7 @@ pub const DOWNLOAD_LIMIT_COND: &str = "downloadLimitCondition";
 pub const UPLOAD_LIMIT_COND: &str = "uploadLimitCondition";
 pub const READ_PROXY_REQ: &str = "readProxyRequest";
 pub const READ_REDIRECT_REQ: &str = "readRedirectRequest";
+pub const READ_DELETED_NEEDLE: &str = "readDeletedNeedle";
 pub const EMPTY_READ_PROXY_LOC: &str = "emptyReadProxyLocaction";
 pub const FAILED_READ_PROXY_REQ: &str = "failedReadProxyRequest";
 

seaweed-volume/src/storage/volume.rs

@@ -18,7 +18,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Condvar, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 use crate::storage::idx;
 use crate::storage::io::read_exact_at;
@@ -1705,6 +1705,10 @@ impl Volume {
         let mut read_size = nv.size;
         if read_size.is_deleted() {
             if read_deleted && !read_size.is_tombstone() {
+                debug!("reading deleted {}", n.id);
+                crate::metrics::HANDLER_COUNTER
+                    .with_label_values(&[crate::metrics::READ_DELETED_NEEDLE])
+                    .inc();
                 read_size = Size(-read_size.0);
             } else {
                 return Err(VolumeError::Deleted);

weed/s3api/s3api_object_handlers_put.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"hash"
 	"io"
+	"math"
 	"net/http"
 	"net/url"
 	"path"
@@ -29,6 +30,8 @@ import (
 	weed_server "github.com/seaweedfs/seaweedfs/weed/server"
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util/constants"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // Object lock validation errors
@@ -992,17 +995,29 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 
 		// If the entry was never created, the uploaded chunks are orphaned and must be deleted.
 		if !entryCreated {
-			orphaned := chunkResult.FileChunks
-			if manifestChunks, _ := filer.SeparateManifestChunks(entry.GetChunks()); len(manifestChunks) > 0 {
-				orphaned = append(manifestChunks, orphaned...)
+			// A transport failure is ambiguous: the filer may have committed the
+			// entry anyway (issue #11366), so a retryable error never deletes the
+			// uploaded chunks — it is only upgraded to success when the write owner
+			// proves the entry landed with these chunks.
+			ambiguous := createErr != nil && filerErrorToS3Error(createErr) == s3err.ErrServiceUnavailable
+			if ambiguous && len(chunkResult.FileChunks) > 0 && s3a.confirmCreateLanded(filePath, bucket, object, entry, chunkResult.FileChunks, finalize) {
+				createCode = s3err.ErrNone
 			}
-			if len(orphaned) > 0 {
-				glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(orphaned))
-				s3a.deleteOrphanedChunks(orphaned)
+			if createCode != s3err.ErrNone && !ambiguous {
+				orphaned := chunkResult.FileChunks
+				if manifestChunks, _ := filer.SeparateManifestChunks(entry.GetChunks()); len(manifestChunks) > 0 {
+					orphaned = append(manifestChunks, orphaned...)
+				}
+				if len(orphaned) > 0 {
+					glog.Warningf("putToFiler: finalization failed, attempting to cleanup %d orphaned chunks", len(orphaned))
+					s3a.deleteOrphanedChunks(orphaned)
+				}
 			}
 		}
 
-		return "", createCode, SSEResponseMetadata{}
+		if createCode != s3err.ErrNone {
+			return "", createCode, SSEResponseMetadata{}
+		}
 	}
 	glog.V(3).Infof("putToFiler: CreateEntry SUCCESS for %s", filePath)
 
@@ -1029,6 +1044,74 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	return etag, s3err.ErrNone, responseMetadata
 }
 
+// confirmCreateLanded checks whether a create that failed ambiguously still
+// committed: the stored entry's resolved chunks must be exactly the uploaded
+// ones. On a match the finalization the error skipped runs under the object
+// write lock, and true reports the write as successful.
+func (s3a *S3ApiServer) confirmCreateLanded(filePath, bucket, object string, entry *filer_pb.Entry, uploaded []*filer_pb.FileChunk, finalize *putFinalize) bool {
+	dir, name := path.Dir(filePath), path.Base(filePath)
+	owner := s3a.routableWriteOwner(bucket, object)
+	confirmed := false
+	// Verify, finalize, and roll back inside one critical section: a concurrent
+	// write to the same key must not slip in between them.
+	s3a.withObjectWriteLock(bucket, object, nil, func() s3err.ErrorCode {
+		existing, lookupErr := s3a.lookupEntryPreferringOwner(owner, dir, name)
+		if lookupErr != nil || existing == nil {
+			return s3err.ErrNone
+		}
+		resolved, _, resolveErr := filer.ResolveChunkManifest(context.Background(), s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient)
+		if resolveErr != nil || !sameFileChunks(resolved, uploaded) {
+			return s3err.ErrNone
+		}
+		glog.Warningf("putToFiler: create entry for %s failed but the entry exists, treating the write as successful", filePath)
+		if finalize == nil || finalize.afterCreate == nil {
+			confirmed = true
+			return s3err.ErrNone
+		}
+		if code := finalize.afterCreate(entry); code != s3err.ErrNone {
+			// Same undo the create path applies when post-create finalization fails.
+			if rbErr := s3a.rmObject(context.Background(), dir, name, true, false); rbErr != nil {
+				glog.Errorf("putToFiler: failed to rollback recovered entry for %s: %v", filePath, rbErr)
+			}
+			return s3err.ErrNone
+		}
+		confirmed = true
+		return s3err.ErrNone
+	})
+	return confirmed
+}
+
+// sameFileChunks reports whether two chunk lists reference the same needles,
+// regardless of order. File id strings are normalized through the parsed Fid so
+// a non-canonical representation cannot masquerade as a different chunk.
+func sameFileChunks(a, b []*filer_pb.FileChunk) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	key := func(c *filer_pb.FileChunk) string {
+		fid := c.GetFid()
+		if fid == nil {
+			fid, _ = filer_pb.ToFileIdObject(c.GetFileIdString())
+		}
+		if fid == nil {
+			return c.GetFileIdString()
+		}
+		return fmt.Sprintf("%d,%x,%x", fid.VolumeId, fid.FileKey, fid.Cookie)
+	}
+	counts := make(map[string]int, len(a))
+	for _, c := range a {
+		counts[key(c)]++
+	}
+	for _, c := range b {
+		k := key(c)
+		if counts[k] == 0 {
+			return false
+		}
+		counts[k]--
+	}
+	return true
+}
+
 // checksumAlgorithmMapping maps algorithm name strings to their enum and header name.
 var checksumAlgorithmMapping = map[string]struct {
 	alg  ChecksumAlgorithm
@@ -1293,13 +1376,23 @@ func filerErrorToS3Error(err error) s3err.ErrorCode {
 		return s3err.ErrAccessDenied
 	}
 
+	// A transport failure leaves the outcome ambiguous — the write may have
+	// been applied anyway — so it must stay retryable, not a permanent 4xx.
+	switch status.Code(err) {
+	case codes.Canceled, codes.DeadlineExceeded, codes.Unavailable:
+		return s3err.ErrServiceUnavailable
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return s3err.ErrServiceUnavailable
+	}
+
 	// Non-filer errors that don't go through CreateEntryResponse — string matching required
 	errString := err.Error()
 	switch {
 	case errString == constants.ErrMsgBadDigest:
 		return s3err.ErrBadDigest
 	case strings.Contains(errString, "context canceled") || strings.Contains(errString, "code = Canceled"):
-		return s3err.ErrInvalidRequest
+		return s3err.ErrServiceUnavailable
 	default:
 		return s3err.ErrInternalError
 	}

weed/s3api/s3api_object_handlers_put_ambiguous_test.go

@@ -0,0 +1,265 @@
+package s3api
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+)
+
+// fakeVolumeServer serves the two volume-server calls putToFiler makes: chunk
+// uploads over HTTP and BatchDelete over gRPC. Deleted fids are recorded so a
+// test can tell whether chunk cleanup ran.
+type fakeVolumeServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	httpAddr string
+	grpcPort uint32
+
+	mu          sync.Mutex
+	deletedFids []string
+}
+
+func (f *fakeVolumeServer) BatchDelete(_ context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	resp := &volume_server_pb.BatchDeleteResponse{}
+	for _, fid := range req.FileIds {
+		f.deletedFids = append(f.deletedFids, fid)
+		resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{FileId: fid, Status: http.StatusAccepted})
+	}
+	return resp, nil
+}
+
+func (f *fakeVolumeServer) deleted() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.deletedFids...)
+}
+
+func startFakeVolumeServer(t *testing.T) *fakeVolumeServer {
+	t.Helper()
+	v := &fakeVolumeServer{}
+	upload := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		io.Copy(io.Discard, r.Body)
+		w.Header().Set("Content-MD5", r.Header.Get("Content-MD5"))
+		w.WriteHeader(http.StatusCreated)
+		io.WriteString(w, `{"size":1}`)
+	}))
+	t.Cleanup(upload.Close)
+	v.httpAddr = strings.TrimPrefix(upload.URL, "http://")
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	grpcSrv := grpc.NewServer()
+	volume_server_pb.RegisterVolumeServerServer(grpcSrv, v)
+	go grpcSrv.Serve(lis)
+	t.Cleanup(grpcSrv.Stop)
+	v.grpcPort = uint32(lis.Addr().(*net.TCPAddr).Port)
+	return v
+}
+
+// ambiguousPutFiler fakes the filer calls putToFiler makes. CreateEntry can
+// apply the write and still return an error — the ambiguous outcome a
+// restarting owner filer produces for issue 11366.
+type ambiguousPutFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	volume *fakeVolumeServer
+
+	mu            sync.Mutex
+	entries       map[string]*filer_pb.Entry
+	apply         bool
+	createErr     error
+	lookupErr     error
+	lookupFailKey string
+	nextKey       uint64
+}
+
+func (f *ambiguousPutFiler) AssignVolume(context.Context, *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.nextKey++
+	return &filer_pb.AssignVolumeResponse{
+		FileId: fmt.Sprintf("3,%016x%08x", f.nextKey, uint32(f.nextKey)),
+		Count:  1,
+		Location: &filer_pb.Location{
+			Url:       f.volume.httpAddr,
+			PublicUrl: f.volume.httpAddr,
+			GrpcPort:  f.volume.grpcPort,
+		},
+	}, nil
+}
+
+func (f *ambiguousPutFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.apply {
+		entry := proto.Clone(req.Entry).(*filer_pb.Entry)
+		filer_pb.BeforeEntrySerialization(entry.Chunks)
+		f.entries[req.Directory+"/"+req.Entry.Name] = entry
+	}
+	if f.createErr != nil {
+		return nil, f.createErr
+	}
+	return &filer_pb.CreateEntryResponse{}, nil
+}
+
+func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.lookupErr != nil && req.Directory+"/"+req.Name == f.lookupFailKey {
+		return nil, f.lookupErr
+	}
+	if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok {
+		out := proto.Clone(entry).(*filer_pb.Entry)
+		filer_pb.AfterEntryDeserialization(out.Chunks)
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: out}, nil
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{}, nil
+}
+
+func (f *ambiguousPutFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
+	resp := &filer_pb.LookupVolumeResponse{LocationsMap: map[string]*filer_pb.Locations{}}
+	for _, vid := range req.VolumeIds {
+		resp.LocationsMap[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{
+			Url:       f.volume.httpAddr,
+			PublicUrl: f.volume.httpAddr,
+			GrpcPort:  f.volume.grpcPort,
+		}}}
+	}
+	return resp, nil
+}
+
+func newPutTestServer(t *testing.T, filerAddr pb.ServerAddress) *S3ApiServer {
+	t.Helper()
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	return &S3ApiServer{
+		option: &S3ApiServerOption{
+			Filers:         []pb.ServerAddress{filerAddr},
+			GrpcDialOption: dialOption,
+			BucketsPath:    "/buckets",
+		},
+		filerClient: wdclient.NewFilerClient([]pb.ServerAddress{filerAddr}, dialOption, ""),
+	}
+}
+
+func putTestObject(t *testing.T, s3a *S3ApiServer) (string, s3err.ErrorCode) {
+	t.Helper()
+	r := httptest.NewRequest(http.MethodPut, "/b/o", nil)
+	etag, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader("hello world"), "b", "o", 1, 0, nil, false, "")
+	return etag, code
+}
+
+// Issue 11366: CreateEntry applied on the filer but the response was lost
+// (owner restarting). Once the entry is confirmed, the write is successful —
+// deleting the chunks would leave the entry pointing at tombstoned needles.
+func TestPutToFilerAmbiguousCreateKeepsChunks(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{
+		volume:    volume,
+		entries:   map[string]*filer_pb.Entry{},
+		apply:     true,
+		createErr: status.Error(codes.Unavailable, "connect: connection refused"),
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+
+	etag, code := putTestObject(t, s3a)
+	if code != s3err.ErrNone {
+		t.Fatalf("putToFiler returned %v, want success once the entry is confirmed on the filer", code)
+	}
+	if etag == "" {
+		t.Fatal("expected an etag")
+	}
+	if deleted := volume.deleted(); len(deleted) != 0 {
+		t.Fatalf("chunks under a live entry were deleted: %v", deleted)
+	}
+}
+
+// A create the filer definitively refused still cleans up the uploaded chunks.
+func TestPutToFilerConfirmedFailureDeletesOrphans(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{
+		volume:    volume,
+		entries:   map[string]*filer_pb.Entry{},
+		apply:     false,
+		createErr: status.Error(codes.Unknown, "create refused"),
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+
+	_, code := putTestObject(t, s3a)
+	if code == s3err.ErrNone {
+		t.Fatal("expected an error when the entry was not created")
+	}
+	if deleted := volume.deleted(); len(deleted) == 0 {
+		t.Fatal("orphaned chunks were not deleted")
+	}
+}
+
+// A stale entry from an earlier object does not prove this PUT landed: the
+// outcome stays unknown, so the new chunks are kept and an error returned.
+func TestPutToFilerAmbiguousCreateWithStaleEntryKeepsChunks(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	stale := &filer_pb.Entry{
+		Name:       "o",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "3,000000000000009900000099", Size: 5}},
+	}
+	filerImpl := &ambiguousPutFiler{
+		volume:    volume,
+		entries:   map[string]*filer_pb.Entry{"/buckets/b/o": stale},
+		apply:     false,
+		createErr: status.Error(codes.Unavailable, "connect: connection refused"),
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+
+	_, code := putTestObject(t, s3a)
+	if code == s3err.ErrNone {
+		t.Fatal("expected an error when the create outcome is unknown")
+	}
+	if deleted := volume.deleted(); len(deleted) != 0 {
+		t.Fatalf("chunks were deleted while the create outcome was unverifiable: %v", deleted)
+	}
+}
+
+// When neither the create nor the lookup can be answered, the outcome stays
+// unknown: keep the chunks (vacuum reclaims orphans) rather than risk deleting
+// chunks a live entry references.
+func TestPutToFilerUnverifiableCreateKeepsChunks(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	unavailable := status.Error(codes.Unavailable, "connect: connection refused")
+	filerImpl := &ambiguousPutFiler{
+		volume:        volume,
+		entries:       map[string]*filer_pb.Entry{},
+		apply:         false,
+		createErr:     unavailable,
+		lookupErr:     unavailable,
+		lookupFailKey: "/buckets/b/o",
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+
+	_, code := putTestObject(t, s3a)
+	if code == s3err.ErrNone {
+		t.Fatal("expected an error when the create outcome is unknown")
+	}
+	if deleted := volume.deleted(); len(deleted) != 0 {
+		t.Fatalf("chunks were deleted while the create outcome was unverifiable: %v", deleted)
+	}
+}

weed/stats/metrics_names.go

@@ -10,6 +10,7 @@ const (
 	UploadLimitCond    = "uploadLimitCondition"
 	ReadProxyReq       = "readProxyRequest"
 	ReadRedirectReq    = "readRedirectRequest"
+	ReadDeletedNeedle  = "readDeletedNeedle"
 	EmptyReadProxyLoc  = "emptyReadProxyLocaction"
 	FailedReadProxyReq = "failedReadProxyRequest"
 

weed/storage/volume_read.go

@@ -35,6 +35,7 @@ func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSize
 	if readSize.IsDeleted() {
 		if readOption != nil && readOption.ReadDeleted && readSize != TombstoneFileSize {
 			glog.V(3).Infof("reading deleted %s", n.String())
+			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ReadDeletedNeedle).Inc()
 			readSize = -readSize
 		} else {
 			return -1, ErrorDeleted