s3: route single-entry object writes to the owner filer, off the DLM (#9629)

s3: route non-versioned object PUT and DELETE off the distributed lock

A non-versioned, non-object-lock object write now goes straight to the key's
owner filer as a single-mutation ObjectTransaction, which serializes it with the
owner's per-path lock and evaluates the precondition, instead of taking a
cluster-wide lock. PUT and DELETE use the object's full path as the lock key, so
a concurrent create and delete of the same key serialize against each other.

The fast path is taken only when the precondition reduces to clauses the filer
can evaluate (existence and a single strong-ETag match); time-based conditions,
ETag lists, weak ETags, post-create hooks, and an unknown owner fall back to the
lock. A routed mutation error other than a failed precondition also falls back,
so the lock path stays the authority for the cases it alone covers.

PrimaryForKey returns "" until the ring view arrives, keeping writes on the lock
until routing is known.

weed/s3api/s3api_object_handlers_copy_part_sse.go

@@ -397,7 +397,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
 	filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID)
 	// Copy-part is an MPU part write under .uploads/<id>/<n>; lifecycle
 	// TTL only applies to the eventual completed object. Pass 0.
-	tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil)
+	tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false)
 	if code != s3err.ErrNone {
 		return "", SSEResponseMetadata{}, code
 	}

weed/s3api/s3api_object_handlers_delete.go

@@ -214,33 +214,59 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 	}
 
 	var deleteResult deleteMutationResult
-	deleteCode := s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
-		return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)
-	}, func() s3err.ErrorCode {
-		if versioningConfigured {
-			result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState)
-			if errCode != s3err.ErrNone {
-				return errCode
+	var deleteCode s3err.ErrorCode
+
+	// Fast path: route the delete to the owner filer under its per-path lock;
+	// routedObjectOwner excludes versioned/object-lock buckets.
+	deleteHandled := false
+	if !versioningConfigured {
+		if cond, condOk := buildDeleteCondition(r); condOk {
+			if owner, ownerOk := s3a.routedObjectOwner(bucket, object); ownerOk {
+				resp, err := s3a.routedDelete(owner, bucket, object, cond)
+				switch {
+				case err != nil:
+					glog.Warningf("DeleteObjectHandler: routed delete to %s failed for %s/%s, falling back to lock: %v", owner, bucket, object, err)
+				case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
+					deleteCode, deleteHandled = s3err.ErrPreconditionFailed, true
+				case resp.Error != "":
+					// Non-precondition error: fall back (the lock path handles cases
+					// the raw delete cannot, e.g. a non-empty directory marker).
+					glog.Warningf("DeleteObjectHandler: routed delete to %s returned %q for %s/%s, falling back to lock", owner, resp.Error, bucket, object)
+				default:
+					deleteCode, deleteHandled = s3err.ErrNone, true
+				}
 			}
-			deleteResult = result
+		}
+	}
+	if !deleteHandled {
+		deleteCode = s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
+			return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)
+		}, func() s3err.ErrorCode {
+			if versioningConfigured {
+				result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState)
+				if errCode != s3err.ErrNone {
+					return errCode
+				}
+				deleteResult = result
+				return s3err.ErrNone
+			}
+
+			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+			if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
+				glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
+				return s3err.ErrAccessDenied
+			}
+
+			if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+				return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
+			}); err != nil {
+				glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err)
+				return s3err.ErrInternalError
+			}
+
 			return s3err.ErrNone
-		}
-
-		governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-		if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
-			glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
-			return s3err.ErrAccessDenied
-		}
-
-		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
-		}); err != nil {
-			glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err)
-			return s3err.ErrInternalError
-		}
-
-		return s3err.ErrNone
-	})
+		})
+	}
 	if deleteCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, deleteCode)
 		return

weed/s3api/s3api_object_handlers_multipart.go

@@ -457,7 +457,7 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 	// transient .uploads/<id>/<n> path, and a part write would otherwise
 	// start the TTL clock before CompleteMultipartUpload ever assembled
 	// the object.
-	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil)
+	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("PutObjectPart: putToFiler failed with error code %v for bucket=%s, object=%s, partNumber=%d",
 			errCode, bucket, object, partID)

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -135,7 +135,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	// fields and boundaries inflates ContentLength relative to the
 	// object body, which would mis-evaluate any size-filtered rule.
 	ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, fileSize)
-	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil)
+	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil, false)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)

weed/s3api/s3api_object_handlers_put.go

@@ -297,7 +297,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			}
 
 			ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, r.ContentLength)
-			etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil)
+			etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil, false)
 
 			if errCode != s3err.ErrNone {
 				s3err.WriteErrorResponse(w, r, errCode)
@@ -359,7 +359,7 @@ func (s3a *S3ApiServer) withObjectWriteLock(bucket, object string, preconditionF
 // pass 0 because their own keys aren't the user-visible object the rule
 // targets and a part write would otherwise bind a TTL clock starting
 // before CompleteMultipartUpload.
-func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
+func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode, uniqueWritePath bool) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
 	// NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy
 	// This eliminates the filer proxy overhead for PUT operations
 	// Note: filePath is now passed directly instead of URL (no parsing needed)
@@ -802,7 +802,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 		}
 		return s3a.checkConditionalHeaders(r, bucket, object)
 	}
-	createCode := s3a.withObjectWriteLock(bucket, object, preconditionFn, func() s3err.ErrorCode {
+	createUnderLock := func() s3err.ErrorCode {
 		createErr = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 			req := &filer_pb.CreateEntryRequest{
 				Directory: path.Dir(filePath),
@@ -831,7 +831,36 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 			}
 		}
 		return s3err.ErrNone
-	})
+	}
+
+	// Route the create to the object's owner filer, whose per-path lock
+	// serializes it, then run afterCreate (e.g. a versioned finalize that routes
+	// itself). Conditional/object-lock/non-reducible cases fall back to the
+	// distributed lock.
+	var createCode s3err.ErrorCode
+	routed := false
+	if owner := s3a.routableWriteOwner(bucket, object); owner != "" {
+		if cond, ok := routeWriteCondition(r, uniqueWritePath); ok {
+			resp, err := s3a.routedPut(owner, filePath, entry, cond)
+			switch {
+			case err != nil:
+				glog.Warningf("putToFiler: routed PUT to %s failed for %s, falling back to lock: %v", owner, filePath, err)
+			case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
+				createCode, routed = s3err.ErrPreconditionFailed, true
+			case resp.Error != "":
+				// Non-precondition mutation error: fall back so the lock path maps it.
+				glog.Warningf("putToFiler: routed PUT to %s returned %q for %s, falling back to lock", owner, resp.Error, filePath)
+			default:
+				entryCreated, routed, createCode = true, true, s3err.ErrNone
+				if afterCreate != nil {
+					createCode = afterCreate(entry)
+				}
+			}
+		}
+	}
+	if !routed {
+		createCode = s3a.withObjectWriteLock(bucket, object, preconditionFn, createUnderLock)
+	}
 	if createCode != s3err.ErrNone {
 		if createErr != nil {
 			glog.Errorf("putToFiler: failed to create entry for %s: %v", filePath, createErr)
@@ -1287,7 +1316,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
 			glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
 		}
 		return s3err.ErrNone
-	})
+	}, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
 		return "", errCode, SSEResponseMetadata{}
@@ -1459,7 +1488,7 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin
 			return s3err.ErrInternalError
 		}
 		return s3err.ErrNone
-	})
+	}, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putVersionedObject: failed to upload version: %v", errCode)
 		return "", "", errCode, SSEResponseMetadata{}

weed/s3api/s3api_object_routed_write.go

@@ -0,0 +1,171 @@
+package s3api
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"path"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// routableWriteOwner returns the owner filer for an object's writes, or "" to
+// keep them on the distributed lock. Versioned and object-lock buckets stay on
+// the lock (handled by later routing PRs); any lookup error also falls back.
+func (s3a *S3ApiServer) routableWriteOwner(bucket, object string) pb.ServerAddress {
+	if object == "" || s3a.objectWriteLockClient == nil {
+		return ""
+	}
+	if configured, err := s3a.isVersioningConfigured(bucket); err != nil || configured {
+		return ""
+	}
+	if locked, err := s3a.isObjectLockEnabled(bucket); err != nil || locked {
+		return ""
+	}
+	return s3a.objectWriteLockClient.PrimaryForKey(fmt.Sprintf("s3.object.write:%s", s3a.toFilerPath(bucket, object)))
+}
+
+// routedObjectOwner resolves the owner for the unversioned DELETE fast path.
+func (s3a *S3ApiServer) routedObjectOwner(bucket, object string) (pb.ServerAddress, bool) {
+	owner := s3a.routableWriteOwner(bucket, object)
+	return owner, owner != ""
+}
+
+// routeWriteCondition reduces the request's conditional headers for a routed
+// create. A unique version path carries no precondition and only routes when the
+// request is unconditional (a conditional versioned write must check the latest,
+// which the lock path does); an overwrite carries the reduced condition.
+func routeWriteCondition(r *http.Request, uniqueWritePath bool) (*filer_pb.WriteCondition, bool) {
+	cond, ok := buildWriteCondition(r)
+	if !ok {
+		return nil, false
+	}
+	if uniqueWritePath && cond != nil {
+		return nil, false
+	}
+	return cond, true
+}
+
+// buildWriteCondition reduces the request's conditional headers to a
+// WriteCondition. ok=false (combined headers, time conditions, ETag lists, weak
+// ETags) keeps gateway-side evaluation under the lock; a nil condition with
+// ok=true means unconditional.
+func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) {
+	headers, errCode := parseConditionalHeaders(r)
+	if errCode != s3err.ErrNone {
+		return nil, false
+	}
+	if !headers.isSet {
+		return nil, true
+	}
+	if !headers.ifModifiedSince.IsZero() || !headers.ifUnmodifiedSince.IsZero() {
+		return nil, false
+	}
+	hasMatch := headers.ifMatch != ""
+	hasNoneMatch := headers.ifNoneMatch != ""
+	switch {
+	case hasMatch && !hasNoneMatch:
+		if headers.ifMatch == "*" {
+			return clause(filer_pb.WriteCondition_IF_EXISTS), true
+		}
+		if etag, single := singleStrongETag(headers.ifMatch); single {
+			return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true
+		}
+		return nil, false
+	case hasNoneMatch && !hasMatch:
+		if headers.ifNoneMatch == "*" {
+			return clause(filer_pb.WriteCondition_IF_NOT_EXISTS), true
+		}
+		if etag, single := singleStrongETag(headers.ifNoneMatch); single {
+			return etagClause(filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, etag), true
+		}
+		return nil, false
+	default:
+		return nil, false
+	}
+}
+
+// buildDeleteCondition reduces a DeleteObject's If-Match header to a condition;
+// DeleteObject honors only If-Match, matching checkDeleteIfMatch.
+func buildDeleteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) {
+	ifMatch := strings.TrimSpace(r.Header.Get(s3_constants.IfMatch))
+	switch {
+	case ifMatch == "":
+		return nil, true
+	case ifMatch == "*":
+		return clause(filer_pb.WriteCondition_IF_EXISTS), true
+	default:
+		if etag, single := singleStrongETag(ifMatch); single {
+			return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true
+		}
+		return nil, false
+	}
+}
+
+func clause(kind filer_pb.WriteCondition_Kind) *filer_pb.WriteCondition {
+	return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: kind}}}
+}
+
+func etagClause(kind filer_pb.WriteCondition_Kind, etag string) *filer_pb.WriteCondition {
+	return &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{Kind: kind, Etags: []string{etag}}}}
+}
+
+// singleStrongETag returns the normalized ETag when v carries exactly one strong
+// ETag, and false for ETag lists or weak ("W/") ETags.
+func singleStrongETag(v string) (string, bool) {
+	v = strings.TrimSpace(v)
+	if strings.Contains(v, ",") {
+		return "", false
+	}
+	if strings.HasPrefix(v, "W/") || strings.HasPrefix(v, "w/") {
+		return "", false
+	}
+	return strings.Trim(v, `"`), true
+}
+
+func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
+	var resp *filer_pb.ObjectTransactionResponse
+	err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		var e error
+		resp, e = client.ObjectTransaction(context.Background(), req)
+		return e
+	})
+	return resp, err
+}
+
+// routedPut writes an object entry as a one-mutation ObjectTransaction on the
+// owner filer. lock_key is the object's full path so the transaction shares the
+// per-path lock with a concurrent create or delete of the same key.
+func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) {
+	return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
+		LockKey:   filePath,
+		Condition: cond,
+		Mutations: []*filer_pb.ObjectMutation{{
+			Type:      filer_pb.ObjectMutation_PUT,
+			Directory: path.Dir(filePath),
+			Entry:     entry,
+		}},
+	})
+}
+
+func (s3a *S3ApiServer) routedDelete(owner pb.ServerAddress, bucket, object string, cond *filer_pb.WriteCondition) (*filer_pb.ObjectTransactionResponse, error) {
+	// NewFullPath normalizes a trailing-slash directory-marker key (e.g. "dir/")
+	// to the entry name "dir", matching deleteUnversionedObjectWithClient.
+	fullpath := util.NewFullPath(s3a.bucketDir(bucket), object)
+	dir, name := fullpath.DirAndName()
+	return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
+		LockKey:   string(fullpath),
+		Condition: cond,
+		Mutations: []*filer_pb.ObjectMutation{{
+			Type:         filer_pb.ObjectMutation_DELETE,
+			Directory:    dir,
+			Name:         name,
+			IsDeleteData: true,
+		}},
+	})
+}

weed/s3api/s3api_object_routed_write_test.go

@@ -0,0 +1,177 @@
+package s3api
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func reqWith(headers map[string]string) *http.Request {
+	r, _ := http.NewRequest(http.MethodPut, "/b/o", nil)
+	for k, v := range headers {
+		r.Header.Set(k, v)
+	}
+	return r
+}
+
+// oneClause returns the single clause of cond, failing if it does not hold
+// exactly one.
+func oneClause(t *testing.T, cond *filer_pb.WriteCondition) *filer_pb.WriteCondition_Clause {
+	t.Helper()
+	if cond == nil {
+		t.Fatal("expected a condition, got nil")
+	}
+	if len(cond.Clauses) != 1 {
+		t.Fatalf("expected 1 clause, got %d", len(cond.Clauses))
+	}
+	return cond.Clauses[0]
+}
+
+func TestBuildWriteCondition(t *testing.T) {
+	t.Run("no headers is unconditional", func(t *testing.T) {
+		cond, ok := buildWriteCondition(reqWith(nil))
+		if !ok || cond != nil {
+			t.Fatalf("want (nil, true), got (%v, %v)", cond, ok)
+		}
+	})
+	t.Run("If-None-Match * to IF_NOT_EXISTS", func(t *testing.T) {
+		cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfNoneMatch: "*"}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_NOT_EXISTS {
+			t.Fatalf("kind = %v", c.Kind)
+		}
+	})
+	t.Run("If-Match * to IF_EXISTS", func(t *testing.T) {
+		cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: "*"}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_EXISTS {
+			t.Fatalf("kind = %v", c.Kind)
+		}
+	})
+	t.Run("If-Match strong etag to IF_ETAG_MATCH", func(t *testing.T) {
+		cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"abc123"`}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		c := oneClause(t, cond)
+		if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" {
+			t.Fatalf("clause = %+v", c)
+		}
+	})
+	t.Run("If-None-Match strong etag to IF_ETAG_NOT_MATCH", func(t *testing.T) {
+		cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfNoneMatch: `"abc123"`}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		c := oneClause(t, cond)
+		if c.Kind != filer_pb.WriteCondition_IF_ETAG_NOT_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" {
+			t.Fatalf("clause = %+v", c)
+		}
+	})
+	t.Run("weak etag falls back", func(t *testing.T) {
+		if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"abc"`})); ok {
+			t.Fatal("weak etag must not take the fast path")
+		}
+	})
+	t.Run("etag list falls back", func(t *testing.T) {
+		if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"a","b"`})); ok {
+			t.Fatal("etag list must not take the fast path")
+		}
+	})
+	t.Run("both match and none-match falls back", func(t *testing.T) {
+		if _, ok := buildWriteCondition(reqWith(map[string]string{
+			s3_constants.IfMatch:     "*",
+			s3_constants.IfNoneMatch: "*",
+		})); ok {
+			t.Fatal("ambiguous combination must not take the fast path")
+		}
+	})
+	t.Run("time-based falls back", func(t *testing.T) {
+		if _, ok := buildWriteCondition(reqWith(map[string]string{
+			"If-Unmodified-Since": "Wed, 21 Oct 2015 07:28:00 GMT",
+		})); ok {
+			t.Fatal("time condition must not take the fast path")
+		}
+	})
+}
+
+func TestBuildDeleteCondition(t *testing.T) {
+	t.Run("no If-Match is unconditional", func(t *testing.T) {
+		cond, ok := buildDeleteCondition(reqWith(nil))
+		if !ok || cond != nil {
+			t.Fatalf("want (nil, true), got (%v, %v)", cond, ok)
+		}
+	})
+	t.Run("If-Match * to IF_EXISTS", func(t *testing.T) {
+		cond, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: "*"}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_EXISTS {
+			t.Fatalf("kind = %v", c.Kind)
+		}
+	})
+	t.Run("If-Match etag to IF_ETAG_MATCH", func(t *testing.T) {
+		cond, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}))
+		if !ok {
+			t.Fatal("want ok")
+		}
+		if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || c.Etags[0] != "e" {
+			t.Fatalf("clause = %+v", c)
+		}
+	})
+	t.Run("weak etag falls back", func(t *testing.T) {
+		if _, ok := buildDeleteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`})); ok {
+			t.Fatal("weak etag must not take the fast path")
+		}
+	})
+}
+
+func TestSingleStrongETag(t *testing.T) {
+	cases := []struct {
+		in     string
+		want   string
+		single bool
+	}{
+		{`"abc"`, "abc", true},
+		{` "abc" `, "abc", true},
+		{`abc`, "abc", true},
+		{`W/"abc"`, "", false},
+		{`w/"abc"`, "", false},
+		{`"a","b"`, "", false},
+	}
+	for _, c := range cases {
+		got, single := singleStrongETag(c.in)
+		if single != c.single || (single && got != c.want) {
+			t.Errorf("singleStrongETag(%q) = (%q, %v), want (%q, %v)", c.in, got, single, c.want, c.single)
+		}
+	}
+}
+
+func TestRouteWriteCondition(t *testing.T) {
+	// Unconditional routes either way.
+	if c, ok := routeWriteCondition(reqWith(nil), false); !ok || c != nil {
+		t.Fatalf("overwrite unconditional: got (%v,%v)", c, ok)
+	}
+	if c, ok := routeWriteCondition(reqWith(nil), true); !ok || c != nil {
+		t.Fatalf("unique unconditional: got (%v,%v)", c, ok)
+	}
+	// An overwrite carries a reducible condition.
+	if c, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}), false); !ok || c == nil {
+		t.Fatalf("overwrite conditional should route: got (%v,%v)", c, ok)
+	}
+	// A conditional unique (versioned) write bails to the lock path.
+	if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}), true); ok {
+		t.Fatal("conditional unique write must not route")
+	}
+	// A non-reducible condition bails regardless.
+	if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`}), false); ok {
+		t.Fatal("weak etag must not route")
+	}
+}