diff --git a/seaweed-volume/src/metrics.rs b/seaweed-volume/src/metrics.rs
index f4774286f..45a52a995 100644
--- a/seaweed-volume/src/metrics.rs
+++ b/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";
 
diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs
index 00f8f3cfe..281bcf5e2 100644
--- a/seaweed-volume/src/storage/volume.rs
+++ b/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);
diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index 472e13895..9cbad8c83 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/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
 	}
diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go
new file mode 100644
index 000000000..9fd094468
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/weed/stats/metrics_names.go b/weed/stats/metrics_names.go
index 956e8df8c..a6fd62948 100644
--- a/weed/stats/metrics_names.go
+++ b/weed/stats/metrics_names.go
@@ -10,6 +10,7 @@ const (
 	UploadLimitCond    = "uploadLimitCondition"
 	ReadProxyReq       = "readProxyRequest"
 	ReadRedirectReq    = "readRedirectRequest"
+	ReadDeletedNeedle  = "readDeletedNeedle"
 	EmptyReadProxyLoc  = "emptyReadProxyLocaction"
 	FailedReadProxyReq = "failedReadProxyRequest"
 
diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go
index fafdcdabd..61fa767fe 100644
--- a/weed/storage/volume_read.go
+++ b/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