diff --git a/docs/en/administration/changelog.md b/docs/en/administration/changelog.md index 32091533d698..7347550acdd4 100644 --- a/docs/en/administration/changelog.md +++ b/docs/en/administration/changelog.md @@ -93,6 +93,8 @@ Example: 103: 1716440760.000000000|UNLINK(1,report.txt,0,false,true):1024|(3,90) ``` +Directory object updates made by the S3 Gateway use a single `SETDIRMARKER(inode,ctime,ctimensec,xattrs)` record. The operation sets the directory's atime and atimensec to zero and applies all listed extended attributes together. The `xattrs` argument is percent-escaped JSON: values are base64-encoded bytes, `null` removes an attribute, and omitted names remain unchanged. External consumers must handle this operation before processing changelogs from gateways that use atomic directory marker updates. The stored inode and xattr formats are unchanged. + ## Notes and limitations {#notes} - The changelog is not a metadata backup. Use [metadata backup](metadata_dump_load.md) for backup and restore. diff --git a/docs/en/guide/gateway.md b/docs/en/guide/gateway.md index 742890f1bdcb..aeb99e97bc57 100644 --- a/docs/en/guide/gateway.md +++ b/docs/en/guide/gateway.md @@ -21,6 +21,14 @@ Common application scenarios for JuiceFS S3 Gateway include: - **Managing files in JuiceFS:** JuiceFS S3 Gateway provides a web-based file manager to manage files in JuiceFS directly from a browser. - **Cluster replication:** In scenarios requiring cross-cluster data replication, JuiceFS S3 Gateway serves as a unified data export for clusters. This avoids cross-region metadata access and enhances data transfer performance. For details, see [Sync across regions using JuiceFS S3 Gateway](../guide/sync.md#sync-across-region). +## Conditional directory objects + +A zero-byte object whose key ends in `/` is represented by a directory with an explicit object marker. Uploading `prefix/child.txt` creates the parent directory, but does not create the S3 object `prefix/`. + +With the default `--head-dir=false`, a `PUT prefix/` or a zero-byte copy to `prefix/` with `If-None-Match: *` can create the marker on this existing directory. The gateway preserves the directory inode and children, and commits the marker and its managed extended attributes in one metadata transaction. Concurrent conditional creators have one winner; subsequent requests return `412 Precondition Failed` without changing the object. + +When `--head-dir` is enabled, implicit directories are exposed as existing objects, so the same conditional request returns `412`. Without the conditional header, a zero-byte copy to `prefix/` creates or overwrites the directory object while preserving existing children. Directory objects must have an empty body. + ## Quick start JuiceFS S3 Gateway enables access to an existing JuiceFS volume. If you do not have one, follow the steps in this [guide](../getting-started/standalone.md) to create a JuiceFS file system. diff --git a/docs/zh_cn/administration/changelog.md b/docs/zh_cn/administration/changelog.md index d0ec6f206f69..829277950567 100644 --- a/docs/zh_cn/administration/changelog.md +++ b/docs/zh_cn/administration/changelog.md @@ -93,6 +93,8 @@ VERSION: UNIX_SECONDS.NANOSECONDS|OPERATION(arguments)[:result]|(SESSION_ID,TXN_ 103: 1716440760.000000000|UNLINK(1,report.txt,0,false,true):1024|(3,90) ``` +S3 网关更新目录对象时使用单条 `SETDIRMARKER(inode,ctime,ctimensec,xattrs)` 记录。该操作将目录的 atime 和 atimensec 设为零，并一并更新列出的扩展属性。`xattrs` 参数为经过百分号转义的 JSON，属性值使用 base64 编码，`null` 表示删除，未列出的属性保持不变。外部消费者在处理使用原子目录标记更新的网关所产生的日志前，需要支持该操作。inode 和扩展属性的存储格式不变。 + ## 使用建议和限制 {#notes} - changelog 不是元数据备份。备份和恢复应使用[元数据备份](metadata_dump_load.md)。 diff --git a/docs/zh_cn/guide/gateway.md b/docs/zh_cn/guide/gateway.md index 03edf61b7e86..2957ada1c493 100644 --- a/docs/zh_cn/guide/gateway.md +++ b/docs/zh_cn/guide/gateway.md @@ -21,6 +21,14 @@ JuiceFS S3 网关的常见的使用场景有： - **管理 JuiceFS 中的文件**：S3 网关提供了一个基于网页的文件管理器，可以在浏览器中管理 JuiceFS 中的文件； - **集群复制**：在跨集群复制数据的场景下，作为集群的统一数据出口，避免跨区访问元数据以提升数据传输性能，详见[「使用 S3 网关进行跨区域数据同步」](../guide/sync.md#sync-across-region) +## 目录对象的条件写入 + +键名以 `/` 结尾的零字节对象，在 JuiceFS 中由带有显式对象标记的目录表示。上传 `prefix/child.txt` 会自动创建父目录，但不会创建 S3 对象 `prefix/`。 + +默认 `--head-dir=false` 时，可以通过带有 `If-None-Match: *` 的 `PUT prefix/` 或向 `prefix/` 复制零字节对象，为已有目录创建对象标记。网关保留目录 inode 和子文件，并在一次元数据事务中提交对象标记及相关扩展属性。并发条件创建只有一个请求成功，后续请求返回 `412 Precondition Failed`，且不会修改已有对象。 + +启用 `--head-dir` 时，隐式目录也被视为已存在的对象，因此上述条件请求返回 `412`。不带条件头时，向 `prefix/` 复制零字节对象会创建或覆盖目录对象，并保留已有子文件。目录对象的请求体必须为空。 + ## 快速开始 启动 S3 网关需要一个已经创建完毕的 JuiceFS 文件系统，如果尚不存在，请参考[文档](../getting-started/standalone.md)来创建。下方假定元数据引擎 URL 为 `redis://localhost:6379/1`。 diff --git a/go.mod b/go.mod index 797b439d7688..250adca31c12 100644 --- a/go.mod +++ b/go.mod @@ -346,7 +346,7 @@ require ( xorm.io/builder v0.3.13 // indirect ) -replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20260515071949-69a6cfc9da65 +replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20260910033240-f2266df40f17 replace github.com/hanwen/go-fuse/v2 v2.1.1-0.20210611132105-24a1dfe6b4f8 => github.com/juicedata/go-fuse/v2 v2.1.1-0.20260819084346-22b3157c2d7f diff --git a/go.sum b/go.sum index 51ff5544eb81..ed3b80a4eba7 100644 --- a/go.sum +++ b/go.sum @@ -505,6 +505,8 @@ github.com/juicedata/huaweicloud-sdk-go-obs v3.22.12-0.20230228031208-386e87b5c0 github.com/juicedata/huaweicloud-sdk-go-obs v3.22.12-0.20230228031208-386e87b5c091+incompatible/go.mod h1:Ukwa8ffRQLV6QRwpqGioPjn2Wnf7TBDA4DbennDOqHE= github.com/juicedata/minio v0.0.0-20260515071949-69a6cfc9da65 h1:u+3ehBnL0r3uQloSRQ6QlY2vffBptFSSN6EeCJqgirE= github.com/juicedata/minio v0.0.0-20260515071949-69a6cfc9da65/go.mod h1:1/4WHQKDOsWA1dd3ADrq9IE/jtFec9MHLy656kIXjNg= +github.com/juicedata/minio v0.0.0-20260910033240-f2266df40f17 h1:PZFzmLxyPGwk/IKgLqr0SuO0P2MI9lq2dlmH0lIX2rQ= +github.com/juicedata/minio v0.0.0-20260910033240-f2266df40f17/go.mod h1:1/4WHQKDOsWA1dd3ADrq9IE/jtFec9MHLy656kIXjNg= github.com/juicedata/mpb/v7 v7.0.4-0.20231024073412-2b8d31be510b h1:0/6suPNZnrOlRlBaU/Bnitu8HiKkkLSzQhHbwQ9AysM= github.com/juicedata/mpb/v7 v7.0.4-0.20231024073412-2b8d31be510b/go.mod h1:NXGsfPGx6G2JssqvEcULtDqUrxuuYs4llpv8W6ZUpzk= github.com/juicedata/xorm v1.4.2-0.20260909084754-f6c8d2b84ec2 h1:j1Ngfz6mtEA8YUhhN3ULLz9UNprl5HLv0SKxsN5cvEY= diff --git a/integration/Makefile b/integration/Makefile index cc15605c4780..729c68911831 100644 --- a/integration/Makefile +++ b/integration/Makefile @@ -2,8 +2,11 @@ all: s3test webdav ioctl s3test: - pip install awscli==1.27.153 - bash s3gateway_test.sh + @awscli_venv=$$(mktemp -d "$${TMPDIR:-/tmp}/juicefs-awscli.XXXXXX"); \ + trap 'rm -rf "$$awscli_venv"' EXIT INT TERM; \ + python3 -m venv "$$awscli_venv"; \ + "$$awscli_venv/bin/pip" install awscli==1.44.73; \ + PATH="$$awscli_venv/bin:$$PATH" bash s3gateway_test.sh webdav: cd /home/travis/.m2/litmus-0.13 ; for i in "basic" "copymove" "http"; do sudo ./$${i} http://127.0.0.1:9009 root 1234; done diff --git a/integration/s3gateway_test.sh b/integration/s3gateway_test.sh index beba9e9298f4..3220b561f466 100755 --- a/integration/s3gateway_test.sh +++ b/integration/s3gateway_test.sh @@ -510,8 +510,8 @@ function test_list_objects() { test_function=${function} out=$($function) rv=$? - output=$(echo "$out") - if [ $rv -eq 0 ] && [ "$output" != "" ]; then + entry_count=$(echo "$out" | jq '((.Contents // []) | length) + ((.CommonPrefixes // []) | length)') + if [ $rv -eq 0 ] && [ "$entry_count" != "0" ]; then rv=1 # since rv is 0, command passed, but didn't return expected value. In this case set the output out="list-objects with prefix is dir failed" @@ -2163,6 +2163,409 @@ function test_object_tagging(){ fi return $rv } + +function test_put_object_if_none_match() { + start_time=$(get_time) + test_function="put-object If-None-Match wildcard" + download_path="/tmp/juicefs-if-none-match-put-$$" + + function="make_bucket" + bucket_name=$(make_bucket) + rv=$? + if [ $rv -ne 0 ]; then + out="${bucket_name}" + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --body ${MINT_DATA_DIR}/datafile-1-kB --bucket ${bucket_name} --key conditional-put --if-none-match '*'" + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-kB" --bucket "${bucket_name}" --key conditional-put --if-none-match '*' 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --body ${MINT_DATA_DIR}/datafile-1-b --bucket ${bucket_name} --key conditional-put --if-none-match '*'" + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-b" --bucket "${bucket_name}" --key conditional-put --if-none-match '*' 2>&1) + status=$? + if [ $status -eq 0 ] || [[ "$out" != *"PreconditionFailed"* ]]; then + rv=1 + out="expected PreconditionFailed for existing object, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api get-object --bucket ${bucket_name} --key conditional-put ${download_path}" + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key conditional-put "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${download_path}" + if [ "$md5rt" != "$HASH_1_KB" ]; then + rv=1 + out="failed conditional PUT changed the existing object" + fi + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --bucket ${bucket_name} --key conditional-dir/ --if-none-match '*'" + out=$(${AWS} s3api put-object --bucket "${bucket_name}" --key conditional-dir/ --if-none-match '*' 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api put-object --bucket "${bucket_name}" --key conditional-dir/ --if-none-match '*' 2>&1) + status=$? + if [ $status -eq 0 ] || [[ "$out" != *"PreconditionFailed"* ]]; then + rv=1 + out="expected PreconditionFailed for existing directory marker, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --body ${MINT_DATA_DIR}/datafile-1-b --bucket ${bucket_name} --key conditional-implicit/child" + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-b" --bucket "${bucket_name}" --key conditional-implicit/child 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --bucket ${bucket_name} --key conditional-implicit/ --if-none-match '*'" + out=$(${AWS} s3api put-object --bucket "${bucket_name}" --key conditional-implicit/ --if-none-match '*' 2>&1) + status=$? + if [ $status -ne 0 ]; then + rv=1 + out="expected successful conditional implicit directory promotion, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api get-object --bucket ${bucket_name} --key conditional-implicit/child ${download_path}" + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key conditional-implicit/child "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${MINT_DATA_DIR}/datafile-1-b" + expected_hash=$md5rt + get_md5 "${download_path}" + if [ "$md5rt" != "$expected_hash" ]; then + rv=1 + out="conditional directory PUT changed the child" + fi + fi + fi + + rm -f "${download_path}" + if [ $rv -eq 0 ]; then + out=$(delete_bucket "${bucket_name}") + rv=$? + else + ${AWS} s3 rb s3://"${bucket_name}" --force > /dev/null 2>&1 + fi + + if [ $rv -eq 0 ]; then + log_success "$(get_duration "$start_time")" "${test_function}" + else + log_failure "$(get_duration "$start_time")" "${function}" "${out}" + fi + return $rv +} + +function test_copy_object_if_none_match() { + start_time=$(get_time) + test_function="copy-object If-None-Match wildcard" + download_path="/tmp/juicefs-if-none-match-copy-$$" + + function="make_bucket" + bucket_name=$(make_bucket) + rv=$? + if [ $rv -ne 0 ]; then + out="${bucket_name}" + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --body ${MINT_DATA_DIR}/datafile-1-kB --bucket ${bucket_name} --key copy-source" + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-kB" --bucket "${bucket_name}" --key copy-source 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api copy-object --bucket ${bucket_name} --key copy-new --copy-source ${bucket_name}/copy-source --if-none-match '*'" + out=$(${AWS} s3api copy-object --bucket "${bucket_name}" --key copy-new --copy-source "${bucket_name}/copy-source" --if-none-match '*' 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key copy-new "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${download_path}" + if [ "$md5rt" != "$HASH_1_KB" ]; then + rv=1 + out="conditional copy produced unexpected data" + fi + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-b" --bucket "${bucket_name}" --key copy-existing 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api copy-object --bucket ${bucket_name} --key copy-existing --copy-source ${bucket_name}/copy-source --if-none-match '*'" + out=$(${AWS} s3api copy-object --bucket "${bucket_name}" --key copy-existing --copy-source "${bucket_name}/copy-source" --if-none-match '*' 2>&1) + status=$? + if [ $status -eq 0 ] || [[ "$out" != *"PreconditionFailed"* ]]; then + rv=1 + out="expected PreconditionFailed for existing copy destination, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key copy-existing "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${MINT_DATA_DIR}/datafile-1-b" + expected_hash=$md5rt + get_md5 "${download_path}" + if [ "$md5rt" != "$expected_hash" ]; then + rv=1 + out="failed conditional copy changed the existing destination" + fi + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --bucket ${bucket_name} --key copy-empty-source" + out=$(${AWS} s3api put-object --bucket "${bucket_name}" --key copy-empty-source 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api put-object --body ${MINT_DATA_DIR}/datafile-1-b --bucket ${bucket_name} --key copy-implicit/child" + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-b" --bucket "${bucket_name}" --key copy-implicit/child 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api copy-object --bucket ${bucket_name} --key copy-implicit/ --copy-source ${bucket_name}/copy-empty-source --if-none-match '*'" + out=$(${AWS} s3api copy-object --bucket "${bucket_name}" --key copy-implicit/ --copy-source "${bucket_name}/copy-empty-source" --if-none-match '*' 2>&1) + status=$? + if [ $status -ne 0 ]; then + rv=1 + out="expected successful conditional copy to an implicit directory, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api get-object --bucket ${bucket_name} --key copy-implicit/child ${download_path}" + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key copy-implicit/child "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${MINT_DATA_DIR}/datafile-1-b" + expected_hash=$md5rt + get_md5 "${download_path}" + if [ "$md5rt" != "$expected_hash" ]; then + rv=1 + out="conditional directory copy changed the child" + fi + fi + fi + + rm -f "${download_path}" + if [ $rv -eq 0 ]; then + out=$(delete_bucket "${bucket_name}") + rv=$? + else + ${AWS} s3 rb s3://"${bucket_name}" --force > /dev/null 2>&1 + fi + + if [ $rv -eq 0 ]; then + log_success "$(get_duration "$start_time")" "${test_function}" + else + log_failure "$(get_duration "$start_time")" "${function}" "${out}" + fi + return $rv +} + +function test_complete_multipart_upload_if_none_match() { + start_time=$(get_time) + test_function="complete-multipart-upload If-None-Match wildcard" + download_path="/tmp/juicefs-if-none-match-multipart-$$" + multipart_file="/tmp/juicefs-if-none-match-multipart-$$.json" + upload_id="" + + function="make_bucket" + bucket_name=$(make_bucket) + rv=$? + if [ $rv -ne 0 ]; then + out="${bucket_name}" + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api put-object --body "${MINT_DATA_DIR}/datafile-1-kB" --bucket "${bucket_name}" --key multipart-target 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api create-multipart-upload --bucket "${bucket_name}" --key multipart-target 2>&1) + rv=$? + upload_id=$(echo "$out" | jq -r .UploadId) + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api upload-part --bucket "${bucket_name}" --key multipart-target --body "${MINT_DATA_DIR}/datafile-5-MB" --upload-id "${upload_id}" --part-number 1 2>&1) + rv=$? + part_etag=$(echo "$out" | jq -r .ETag) + echo "{\"Parts\":[{\"ETag\":${part_etag},\"PartNumber\":1}]}" > "${multipart_file}" + fi + + if [ $rv -eq 0 ]; then + function="${AWS} s3api complete-multipart-upload --multipart-upload file://${multipart_file} --bucket ${bucket_name} --key multipart-target --upload-id ${upload_id} --if-none-match '*'" + out=$(${AWS} s3api complete-multipart-upload --multipart-upload "file://${multipart_file}" --bucket "${bucket_name}" --key multipart-target --upload-id "${upload_id}" --if-none-match '*' 2>&1) + status=$? + if [ $status -eq 0 ] || [[ "$out" != *"PreconditionFailed"* ]]; then + rv=1 + out="expected PreconditionFailed for existing multipart destination, got: ${out}" + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key multipart-target "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${download_path}" + if [ "$md5rt" != "$HASH_1_KB" ]; then + rv=1 + out="failed conditional multipart completion changed the existing object" + fi + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api delete-object --bucket "${bucket_name}" --key multipart-target 2>&1) + rv=$? + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api complete-multipart-upload --multipart-upload "file://${multipart_file}" --bucket "${bucket_name}" --key multipart-target --upload-id "${upload_id}" --if-none-match '*' 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + upload_id="" + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key multipart-target "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${MINT_DATA_DIR}/datafile-5-MB" + expected_hash=$md5rt + get_md5 "${download_path}" + if [ "$md5rt" != "$expected_hash" ]; then + rv=1 + out="retried conditional multipart completion produced unexpected data" + fi + fi + fi + + rm -f "${download_path}" "${multipart_file}" + if [ $rv -ne 0 ] && [ -n "$upload_id" ]; then + ${AWS} s3api abort-multipart-upload --bucket "${bucket_name}" --key multipart-target --upload-id "${upload_id}" > /dev/null 2>&1 + fi + if [ $rv -eq 0 ]; then + out=$(delete_bucket "${bucket_name}") + rv=$? + else + ${AWS} s3 rb s3://"${bucket_name}" --force > /dev/null 2>&1 + fi + + if [ $rv -eq 0 ]; then + log_success "$(get_duration "$start_time")" "${test_function}" + else + log_failure "$(get_duration "$start_time")" "${function}" "${out}" + fi + return $rv +} + +function test_concurrent_put_object_if_none_match() { + start_time=$(get_time) + test_function="concurrent put-object If-None-Match wildcard" + result_dir="/tmp/juicefs-if-none-match-concurrent-$$" + download_path="/tmp/juicefs-if-none-match-concurrent-object-$$" + + function="make_bucket" + bucket_name=$(make_bucket) + rv=$? + if [ $rv -ne 0 ]; then + out="${bucket_name}" + fi + + if [ $rv -eq 0 ]; then + mkdir -p "${result_dir}" + for index in $(seq 1 8); do + if [ $((index % 2)) -eq 0 ]; then + body="${MINT_DATA_DIR}/datafile-1-kB" + else + body="${MINT_DATA_DIR}/datafile-1-b" + fi + echo "${body}" > "${result_dir}/${index}.body" + ( + ${AWS} s3api put-object --body "${body}" --bucket "${bucket_name}" --key concurrent-put --if-none-match '*' > "${result_dir}/${index}.out" 2>&1 + echo "$?" > "${result_dir}/${index}.status" + ) & + done + wait + + success_count=0 + winner_body="" + for index in $(seq 1 8); do + status=$(cat "${result_dir}/${index}.status") + response=$(cat "${result_dir}/${index}.out") + if [ "$status" -eq 0 ]; then + success_count=$((success_count + 1)) + winner_body=$(cat "${result_dir}/${index}.body") + elif [[ "$response" != *"PreconditionFailed"* && "$response" != *"ConditionalRequestConflict"* ]]; then + rv=1 + out="unexpected concurrent conditional PUT failure: ${response}" + break + fi + done + + if [ $rv -eq 0 ] && [ "$success_count" -ne 1 ]; then + rv=1 + out="expected exactly one successful concurrent conditional PUT, got ${success_count}" + fi + fi + + if [ $rv -eq 0 ]; then + out=$(${AWS} s3api get-object --bucket "${bucket_name}" --key concurrent-put "${download_path}" 2>&1) + rv=$? + if [ $rv -eq 0 ]; then + get_md5 "${winner_body}" + expected_hash=$md5rt + get_md5 "${download_path}" + if [ "$md5rt" != "$expected_hash" ]; then + rv=1 + out="stored object does not belong to the successful conditional writer" + fi + fi + fi + + rm -rf "${result_dir}" + rm -f "${download_path}" + if [ $rv -eq 0 ]; then + out=$(delete_bucket "${bucket_name}") + rv=$? + else + ${AWS} s3 rb s3://"${bucket_name}" --force > /dev/null 2>&1 + fi + + if [ $rv -eq 0 ]; then + log_success "$(get_duration "$start_time")" "${test_function}" + else + log_failure "$(get_duration "$start_time")" "${function}" "${out}" + fi + return $rv +} + # main handler for all the tests. main() { # Success tests @@ -2194,6 +2597,10 @@ main() { # test_worm_bucket && \ # test_legal_hold test_get_object_error && \ + test_put_object_if_none_match && \ + test_copy_object_if_none_match && \ + test_complete_multipart_upload_if_none_match && \ + test_concurrent_put_object_if_none_match && \ test_object_tagging return $? } diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index bf44d667b0b7..81e548008d13 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -64,6 +64,20 @@ var logger = utils.GetLogger("juicefs") var bucketLockOwner atomic.Uint64 var bucketLockTimeout = minio.NewDynamicTimeout(2*time.Minute, 1*time.Minute) +func isExplicitDirectoryMarker(attr *meta.Attr) bool { + return attr.Atime*1000+int64(attr.Atimensec/1e6) == 0 +} + +func shouldLoadObjectETag(isDir bool, attr *meta.Attr) bool { + return !isDir || attr != nil && isExplicitDirectoryMarker(attr) +} + +type objectXattr struct { + name string + value []byte + remove bool +} + type Config struct { MultiBucket bool Bucket string @@ -381,7 +395,7 @@ func (n *jfsObjects) listDirFactory() minio.ListDirFunc { } defer f.Close(mctx) if !n.gConf.HideDir { - if fi, _ := f.Stat(); fi.(*fs.FileStat).Atime() == 0 && prefixEntry == "" { + if fi, _ := f.Stat(); isExplicitDirectoryMarker(fi.(*fs.FileStat).Attr()) && prefixEntry == "" { entries = append(entries, &minio.Entry{Name: ""}) } } @@ -439,20 +453,20 @@ func (n *jfsObjects) ListObjects(ctx context.Context, bucket, prefix, marker, de getObjectInfo := func(ctx context.Context, bucket, object string, fi_ any) (obj minio.ObjectInfo, err error) { var eno syscall.Errno var info *minio.ObjectInfo + var fileStat *fs.FileStat if fi_ == nil { - var fi *fs.FileStat - fi, eno = n.fs.Stat(mctx, n.path(bucket, object)) + fileStat, eno = n.fs.Stat(mctx, n.path(bucket, object)) if eno == 0 { - size := fi.Size() - if fi.IsDir() { + size := fileStat.Size() + if fileStat.IsDir() { size = 0 } info = &minio.ObjectInfo{ Bucket: bucket, - ModTime: fi.ModTime(), + ModTime: fileStat.ModTime(), Size: size, - IsDir: fi.IsDir(), - AccTime: fi.ModTime(), + IsDir: fileStat.IsDir(), + AccTime: fileStat.ModTime(), IsLatest: true, } } @@ -471,17 +485,17 @@ func (n *jfsObjects) ListObjects(ctx context.Context, bucket, prefix, marker, de eno = 0 } } else { - fi := fi_.(*fs.FileStat) + fileStat = fi_.(*fs.FileStat) info = &minio.ObjectInfo{ Bucket: bucket, - Name: fi.Name(), - ModTime: fi.ModTime(), - Size: fi.Size(), - IsDir: fi.IsDir(), - AccTime: fi.ModTime(), + Name: fileStat.Name(), + ModTime: fileStat.ModTime(), + Size: fileStat.Size(), + IsDir: fileStat.IsDir(), + AccTime: fileStat.ModTime(), IsLatest: true, } - if fi.IsDir() { + if fileStat.IsDir() { info.Size = 0 } } @@ -490,7 +504,11 @@ func (n *jfsObjects) ListObjects(ctx context.Context, bucket, prefix, marker, de return obj, jfsToObjectErr(ctx, eno, bucket, object) } info.Name = object - if n.gConf.KeepEtag && !strings.HasSuffix(object, sep) { + var attr *meta.Attr + if fileStat != nil { + attr = fileStat.Attr() + } + if n.gConf.KeepEtag && shouldLoadObjectETag(info.IsDir, attr) { etag, _ := n.fs.GetXattr(mctx, n.path(bucket, object), s3Etag) info.ETag = string(etag) } @@ -677,7 +695,31 @@ func (n *jfsObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBu dst := n.path(dstBucket, dstObject) src := n.path(srcBucket, srcObject) + if strings.HasSuffix(dstObject, sep) { + if srcInfo.Size > 0 { + return info, minio.ObjectExistsAsDirectory{ + Bucket: dstBucket, + Object: dstObject, + Err: syscall.EEXIST, + } + } + var etag []byte + var xattrs []objectXattr + if xattrs, etag, err = n.copyDirectoryObjectXattrs(src, srcInfo); err != nil { + return info, err + } + if err = n.putDirectoryObject(ctx, dstBucket, dst, dstOpts.IfNoneMatch, xattrs); err != nil { + return info, jfsToObjectErr(ctx, err, dstBucket, dstObject) + } + info, err = n.GetObjectInfo(ctx, dstBucket, dstObject, minio.ObjectOptions{}) + info.ETag = string(etag) + return info, err + } + if minio.IsStringEqual(src, dst) { + if dstOpts.IfNoneMatch { + return info, minio.PreConditionFailed{} + } // if we copy the same object for set metadata err = n.setObjMeta(dst, srcInfo.UserDefined) if err != nil { @@ -740,14 +782,21 @@ func (n *jfsObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBu return } - eno = n.fs.Rename(mctx, tmp, dst, 0) + renameFlags := uint32(0) + if dstOpts.IfNoneMatch { + renameFlags = meta.RenameNoReplace + } + eno = n.fs.Rename(mctx, tmp, dst, renameFlags) if eno == syscall.ENOENT { if err = n.mkdirAllInBucket(ctx, dstBucket, path.Dir(dst)); err != nil { logger.Errorf("mkdirAll %s: %s", path.Dir(dst), err) err = n.objectCommitErr(ctx, err, dstBucket, dstObject) return } - eno = n.fs.Rename(mctx, tmp, dst, 0) + eno = n.fs.Rename(mctx, tmp, dst, renameFlags) + } + if dstOpts.IfNoneMatch && eno == syscall.EEXIST { + return info, minio.PreConditionFailed{} } if eno != 0 { err = n.objectCommitErr(ctx, eno, dstBucket, dstObject) @@ -787,7 +836,7 @@ func (n *jfsObjects) GetObjectInfo(ctx context.Context, bucket, object string, o // put /dir1/key1; head /dir1 return 404; head /dir1/ return 404; head /dir1/key1 return 200 // put /dir1/key1/; head /dir1/key1 return 404; head /dir1/key1/ return 200 var isObject bool - if strings.HasSuffix(object, sep) && fi.IsDir() && fi.Atime() == 0 { + if strings.HasSuffix(object, sep) && fi.IsDir() && isExplicitDirectoryMarker(fi.Attr()) { isObject = true } else if !strings.HasSuffix(object, sep) && !fi.IsDir() { isObject = true @@ -797,7 +846,7 @@ func (n *jfsObjects) GetObjectInfo(ctx context.Context, bucket, object string, o return } var etag []byte - if n.gConf.KeepEtag && !fi.IsDir() { + if n.gConf.KeepEtag && shouldLoadObjectETag(fi.IsDir(), fi.Attr()) { etag, _ = n.fs.GetXattr(mctx, n.path(bucket, object), s3Etag) } size := fi.Size() @@ -937,7 +986,11 @@ func (n *jfsObjects) putObject(ctx context.Context, bucket, object string, r *mi return } - eno = n.fs.Rename(mctx, tmpname, object, 0) + renameFlags := uint32(0) + if opts.IfNoneMatch { + renameFlags = meta.RenameNoReplace + } + eno = n.fs.Rename(mctx, tmpname, object, renameFlags) if eno == syscall.ENOENT { if strings.HasPrefix(object, sep+metaBucket+sep) { err = n.objectCommitErr(ctx, eno, bucket, object, path.Base(path.Dir(object))) @@ -948,7 +1001,10 @@ func (n *jfsObjects) putObject(ctx context.Context, bucket, object string, r *mi err = n.objectCommitErr(ctx, err, bucket, object) return } - eno = n.fs.Rename(mctx, tmpname, object, 0) + eno = n.fs.Rename(mctx, tmpname, object, renameFlags) + } + if opts.IfNoneMatch && eno == syscall.EEXIST { + return nil, minio.PreConditionFailed{} } if eno != 0 { err = n.objectCommitErr(ctx, eno, bucket, object) @@ -956,6 +1012,135 @@ func (n *jfsObjects) putObject(ctx context.Context, bucket, object string, r *mi return } +func (n *jfsObjects) applyObjectXattrs(ctx meta.Context, inode meta.Ino, xattrs []objectXattr) error { + for _, xattr := range xattrs { + var eno syscall.Errno + if xattr.remove { + eno = n.fs.Meta().RemoveXattr(ctx, inode, xattr.name) + if eno == meta.ENOATTR { + eno = 0 + } + } else { + eno = n.fs.Meta().SetXattr(ctx, inode, xattr.name, xattr.value, 0) + } + if eno != 0 { + return eno + } + } + return nil +} + +func (n *jfsObjects) publishNewDirectoryObject(ctx context.Context, bucket, directoryPath string, xattrs []objectXattr) error { + uuid := minio.MustGetUUID() + tmp := n.tpath(bucket, "tmp", uuid[:subDirPrefix], uuid) + if err := n.mkdirAll(ctx, path.Dir(tmp)); err != nil { + return err + } + if eno := n.fs.Mkdir(mctx, tmp, 0777, n.gConf.Umask); eno != 0 { + return eno + } + published := false + defer func() { + if !published { + _ = n.fs.Delete(mctx, tmp) + } + }() + fi, eno := n.fs.Stat(mctx, tmp) + if eno != 0 { + return eno + } + if err := n.applyObjectXattrs(mctx, fi.Inode(), xattrs); err != nil { + return err + } + attr := meta.Attr{Atime: 0, Atimensec: 0} + if eno = n.fs.Meta().SetAttr(mctx, fi.Inode(), meta.SetAttrAtime, 0, &attr); eno != 0 { + return eno + } + n.fs.InvalidateAttr(fi.Inode()) + // Publish the marker with its attributes already set, leaving any existing + // destination untouched. + eno = n.fs.Rename(mctx, tmp, directoryPath, meta.RenameNoReplace) + if eno != 0 { + return eno + } + published = true + return nil +} + +func (n *jfsObjects) putDirectoryObject(ctx context.Context, bucket, directoryPath string, ifNoneMatch bool, xattrs []objectXattr) error { + directoryPath = strings.TrimSuffix(directoryPath, sep) + parentPath := path.Dir(directoryPath) + if err := n.mkdirAllInBucket(ctx, bucket, parentPath); err != nil { + return err + } + + parent, eno := n.fs.Stat(mctx, parentPath) + if eno != 0 { + return eno + } + + requestCtx := meta.WrapWithCancel(ctx, mctx.Pid(), mctx.Uid(), mctx.Gids()) + defer requestCtx.Cancel() + name := path.Base(directoryPath) + var inode meta.Ino + var attr meta.Attr + eno = n.fs.Meta().Lookup(requestCtx, parent.Inode(), name, &inode, &attr, true) + if eno == syscall.ENOENT { + publishErr := n.publishNewDirectoryObject(ctx, bucket, directoryPath, xattrs) + if publishErr == nil { + return nil + } + if errors.Is(publishErr, syscall.EEXIST) { + eno = n.fs.Meta().Lookup(requestCtx, parent.Inode(), name, &inode, &attr, true) + } else { + return publishErr + } + } + if eno != 0 { + return eno + } + if attr.Typ != meta.TypeDirectory { + if ifNoneMatch { + return minio.PreConditionFailed{} + } + return fmt.Errorf("%s is not directory", directoryPath) + } + if ifNoneMatch && (isExplicitDirectoryMarker(&attr) || n.gConf.HeadDir) { + return minio.PreConditionFailed{} + } + values := make(map[string][]byte, len(xattrs)) + for _, xattr := range xattrs { + if xattr.remove { + values[xattr.name] = nil + } else { + values[xattr.name] = xattr.value + } + } + eno = n.fs.Meta().SetDirMarker(requestCtx, parent.Inode(), name, inode, ifNoneMatch, values, &attr) + if ifNoneMatch && eno == syscall.EEXIST { + return minio.PreConditionFailed{} + } + if eno != 0 { + return eno + } + n.fs.InvalidateAttr(inode) + return nil +} + +func (n *jfsObjects) emptyDirectoryObjectXattrs() []objectXattr { + var xattrs []objectXattr + if n.gConf.KeepEtag { + xattrs = append(xattrs, objectXattr{name: s3Etag, remove: true}) + } + if n.gConf.ObjTag { + xattrs = append(xattrs, objectXattr{name: s3Tags, remove: true}) + } + if n.gConf.ObjMeta { + xattrs = append(xattrs, objectXattr{name: s3Meta, remove: true}) + } + return xattrs +} + func (n *jfsObjects) PutObject(ctx context.Context, bucket string, object string, r *minio.PutObjReader, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) { if err = n.checkBucket(ctx, bucket); err != nil { return @@ -966,10 +1151,6 @@ func (n *jfsObjects) PutObject(ctx context.Context, bucket string, object string var eno syscall.Errno p := n.path(bucket, object) if strings.HasSuffix(object, sep) { - if err = n.mkdirAllInBucket(ctx, bucket, p); err != nil { - err = n.objectCommitErr(ctx, err, bucket, object) - return - } if r.Size() > 0 { err = minio.ObjectExistsAsDirectory{ Bucket: bucket, @@ -978,11 +1159,13 @@ func (n *jfsObjects) PutObject(ctx context.Context, bucket string, object string } return } - // if the put object is a directory, set its atime to 0 - n.setFileAtime(p, 0) + if err = n.putDirectoryObject(ctx, bucket, p, opts.IfNoneMatch, n.emptyDirectoryObjectXattrs()); err != nil { + err = n.objectCommitErr(ctx, err, bucket, object) + return + } fi, eno = n.fs.Stat(mctx, p) if eno != 0 { - return objInfo, jfsToObjectErr(ctx, eno, bucket, object) + return objInfo, n.objectCommitErr(ctx, eno, bucket, object) } } else { if fi, err = n.putObject(ctx, bucket, p, r, opts, func(tmpName string) { @@ -1070,6 +1253,46 @@ var s3UserControlledSystemMeta = []string{ "content-type", } +func objectMetadataValue(metadata map[string]string) ([]byte, error) { + objectMetadata := make(map[string]string) + for k, v := range metadata { + k = strings.ToLower(k) + if strings.HasPrefix(k, amzMeta) { + objectMetadata[k] = v + } else { + for _, systemMetaKey := range s3UserControlledSystemMeta { + if k == systemMetaKey { + objectMetadata[k] = v + break + } + } + } + } + if len(objectMetadata) == 0 { + return nil, nil + } + return json.Marshal(objectMetadata) +} + +func (n *jfsObjects) copyDirectoryObjectXattrs(src string, srcInfo minio.ObjectInfo) (xattrs []objectXattr, etag []byte, err error) { + if n.gConf.KeepEtag { + etag, _ = n.fs.GetXattr(mctx, src, s3Etag) + xattrs = append(xattrs, objectXattr{name: s3Etag, value: etag, remove: len(etag) == 0}) + } + if n.gConf.ObjTag { + tagStr := srcInfo.UserDefined[xhttp.AmzObjectTagging] + xattrs = append(xattrs, objectXattr{name: s3Tags, value: []byte(tagStr), remove: tagStr == ""}) + } + if n.gConf.ObjMeta { + var metadataValue []byte + if metadataValue, err = objectMetadataValue(srcInfo.UserDefined); err != nil { + return nil, nil, err + } + xattrs = append(xattrs, objectXattr{name: s3Meta, value: metadataValue, remove: len(metadataValue) == 0}) + } + return xattrs, etag, nil +} + func (n *jfsObjects) getObjMeta(p string) (objMeta map[string]string, err error) { if n.gConf.ObjMeta { var errno syscall.Errno @@ -1089,25 +1312,11 @@ func (n *jfsObjects) getObjMeta(p string) (objMeta map[string]string, err error) func (n *jfsObjects) setObjMeta(p string, metadata map[string]string) error { if n.gConf.ObjMeta && metadata != nil { - meta := make(map[string]string) - for k, v := range metadata { - k = strings.ToLower(k) - if strings.HasPrefix(k, amzMeta) { - meta[k] = v - } else { - for _, systemMetaKey := range s3UserControlledSystemMeta { - if k == systemMetaKey { - meta[k] = v - break - } - } - } + s3MetadataValue, err := objectMetadataValue(metadata) + if err != nil { + return err } - if len(meta) > 0 { - s3MetadataValue, err := json.Marshal(meta) - if err != nil { - return err - } + if len(s3MetadataValue) > 0 { if eno := n.fs.SetXattr(mctx, p, s3Meta, s3MetadataValue, 0); eno != 0 { logger.Errorf("set object metadata error, path: %s,value: %s error: %s", p, string(s3Meta), eno) } @@ -1406,7 +1615,11 @@ func (n *jfsObjects) CompleteMultipartUpload(ctx context.Context, bucket, object } name := n.path(bucket, object) - eno = n.fs.Rename(mctx, tmp, name, 0) + renameFlags := uint32(0) + if opts.IfNoneMatch { + renameFlags = meta.RenameNoReplace + } + eno = n.fs.Rename(mctx, tmp, name, renameFlags) if eno == syscall.ENOENT { if err = n.mkdirAllInBucket(ctx, bucket, path.Dir(name)); err != nil { logger.Errorf("mkdirAll %s: %s", path.Dir(name), err) @@ -1414,7 +1627,11 @@ func (n *jfsObjects) CompleteMultipartUpload(ctx context.Context, bucket, object err = n.objectCommitErr(ctx, err, bucket, object, uploadID) return } - eno = n.fs.Rename(mctx, tmp, name, 0) + eno = n.fs.Rename(mctx, tmp, name, renameFlags) + } + if opts.IfNoneMatch && eno == syscall.EEXIST { + _ = n.fs.Delete(mctx, tmp) + return objInfo, minio.PreConditionFailed{} } if eno != 0 { _ = n.fs.Delete(mctx, tmp) diff --git a/pkg/gateway/gateway_test.go b/pkg/gateway/gateway_test.go index e157a745f2a8..842f79d393a3 100644 --- a/pkg/gateway/gateway_test.go +++ b/pkg/gateway/gateway_test.go @@ -22,6 +22,7 @@ import ( "crypto/md5" "encoding/hex" "errors" + "fmt" "io" "os" "path" @@ -37,6 +38,7 @@ import ( "github.com/juicedata/juicefs/pkg/object" "github.com/juicedata/juicefs/pkg/vfs" minio "github.com/minio/minio/cmd" + xhttp "github.com/minio/minio/cmd/http" miniohash "github.com/minio/minio/pkg/hash" ) @@ -681,6 +683,19 @@ func TestMkdirAllInBucket(t *testing.T) { if _, eno := jfs.Stat(mctx, jfsObj.path(missingBucket)); !fs.IsNotExist(eno) { t.Fatalf("missing bucket was recreated: %s", eno) } + + // A bucket can disappear after PutObject checks it, before the marker is committed. + for _, ifNoneMatch := range []bool{false, true} { + for _, object := range []string{"marker/", "dir/marker/"} { + err := jfsObj.putDirectoryObject(ctx, missingBucket, jfsObj.path(missingBucket, object), ifNoneMatch, nil) + if err == nil || !fs.IsNotExist(err) { + t.Fatalf("directory write under missing bucket (conditional=%t, object=%s): expected ENOENT, got %v", ifNoneMatch, object, err) + } + if _, eno := jfs.Stat(mctx, jfsObj.path(missingBucket)); !fs.IsNotExist(eno) { + t.Fatalf("directory write recreated missing bucket: %s", eno) + } + } + } } func createTestFile(t *testing.T, jfs *fs.FileSystem, name string) { @@ -694,6 +709,940 @@ func createTestFile(t *testing.T, jfs *fs.FileSystem, name string) { } } +func newPutObjectReader(t *testing.T, data []byte) *minio.PutObjReader { + t.Helper() + reader, err := miniohash.NewReader(bytes.NewReader(data), int64(len(data)), "", "", int64(len(data))) + if err != nil { + t.Fatalf("create put object reader: %s", err) + } + return minio.NewPutObjReader(reader) +} + +func readGatewayObject(t *testing.T, gateway *jfsObjects, bucket, object string) []byte { + t.Helper() + reader, err := gateway.GetObjectNInfo(context.Background(), bucket, object, nil, nil, minio.LockType(0), minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get %s: %s", object, err) + } + defer reader.Close() + data, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("read %s: %s", object, err) + } + return data +} + +func assertPreconditionFailed(t *testing.T, err error) { + t.Helper() + var preconditionFailed minio.PreConditionFailed + if !errors.As(err, &preconditionFailed) { + t.Fatalf("expected PreConditionFailed, got %T: %v", err, err) + } +} + +func TestPutObjectIfNoneMatch(t *testing.T) { + t.Run("existing object remains unchanged", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + original := []byte("original") + if _, err := gateway.PutObject(ctx, bucket, "object", newPutObjectReader(t, original), minio.ObjectOptions{}); err != nil { + t.Fatalf("create original object: %s", err) + } + + _, err := gateway.PutObject(ctx, bucket, "object", newPutObjectReader(t, []byte("replacement")), minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + if got := readGatewayObject(t, gateway, bucket, "object"); !bytes.Equal(got, original) { + t.Fatalf("existing object changed: got %q, want %q", got, original) + } + }) + + t.Run("concurrent creates have one winner", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + const writers = 16 + start := make(chan struct{}) + type result struct { + payload []byte + err error + } + results := make(chan result, writers) + readers := make([]*minio.PutObjReader, writers) + var waitGroup sync.WaitGroup + for i := 0; i < writers; i++ { + payload := []byte(fmt.Sprintf("writer-%02d", i)) + readers[i] = newPutObjectReader(t, payload) + waitGroup.Add(1) + go func(payload []byte, reader *minio.PutObjReader) { + defer waitGroup.Done() + <-start + _, err := gateway.PutObject(ctx, bucket, "race", reader, minio.ObjectOptions{IfNoneMatch: true}) + results <- result{payload: payload, err: err} + }(payload, readers[i]) + } + close(start) + waitGroup.Wait() + close(results) + + succeeded := 0 + failed := 0 + var successfulPayload []byte + for result := range results { + if result.err == nil { + succeeded++ + successfulPayload = result.payload + continue + } + var preconditionFailed minio.PreConditionFailed + if errors.As(result.err, &preconditionFailed) { + failed++ + continue + } + t.Fatalf("unexpected concurrent PUT error: %T: %v", result.err, result.err) + } + if succeeded != 1 || failed != writers-1 { + t.Fatalf("conditional create results: succeeded=%d failed=%d", succeeded, failed) + } + if stored := readGatewayObject(t, gateway, bucket, "race"); !bytes.Equal(stored, successfulPayload) { + t.Fatalf("stored payload %q does not match successful writer %q", stored, successfulPayload) + } + }) + + t.Run("directory marker can only be created once", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + opts := minio.ObjectOptions{IfNoneMatch: true} + if _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), opts); err != nil { + t.Fatalf("conditionally create directory marker: %s", err) + } + _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), opts) + assertPreconditionFailed(t, err) + }) + + t.Run("directory marker writes can proceed while the parent is locked", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{}) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + parent, eno := jfs.Stat(mctx, gateway.path(bucket, "")) + if eno != 0 { + t.Fatalf("stat parent directory: %s", eno) + } + const owner = ^uint64(0) + if eno = jfs.Meta().Flock(mctx, parent.Inode(), owner, meta.F_WRLCK, false); eno != 0 { + t.Fatalf("lock parent directory: %s", eno) + } + defer func() { + if eno := jfs.Meta().Flock(mctx, parent.Inode(), owner, meta.F_UNLCK, false); eno != 0 { + t.Errorf("unlock parent directory: %s", eno) + } + }() + + for _, conditional := range []bool{true, false} { + if _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{IfNoneMatch: conditional}); err != nil { + t.Fatalf("write marker with IfNoneMatch=%t while parent is locked: %s", conditional, err) + } + } + assertHeadObject(t, gateway, bucket, "prefix/", true) + }) + + t.Run("head dir concurrent marker creates have one winner", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{HeadDir: true}) + ctx := context.Background() + const writers = 8 + start := make(chan struct{}) + results := make(chan error, writers) + var waitGroup sync.WaitGroup + for i := 0; i < writers; i++ { + reader := newPutObjectReader(t, nil) + waitGroup.Add(1) + go func(reader *minio.PutObjReader) { + defer waitGroup.Done() + <-start + _, err := gateway.PutObject(ctx, bucket, "prefix/", reader, minio.ObjectOptions{IfNoneMatch: true}) + results <- err + }(reader) + } + close(start) + waitGroup.Wait() + close(results) + + succeeded := 0 + failed := 0 + for err := range results { + if err == nil { + succeeded++ + continue + } + var preconditionFailed minio.PreConditionFailed + if errors.As(err, &preconditionFailed) { + failed++ + continue + } + t.Fatalf("unexpected concurrent directory PUT error: %T: %v", err, err) + } + if succeeded != 1 || failed != writers-1 { + t.Fatalf("conditional directory results: succeeded=%d failed=%d", succeeded, failed) + } + assertHeadObject(t, gateway, bucket, "prefix/", true) + }) + + t.Run("implicit directory is promoted without replacing children", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + directoryPath := gateway.path(bucket, "prefix") + fi, eno := jfs.Stat(mctx, directoryPath) + if eno != 0 { + t.Fatalf("stat implicit directory: %s", eno) + } + attr := meta.Attr{Atime: 123, Atimensec: 456000000} + if eno = jfs.Meta().SetAttr(mctx, fi.Inode(), meta.SetAttrAtime, 0, &attr); eno != 0 { + t.Fatalf("set implicit directory atime: %s", eno) + } + jfs.InvalidateAttr(fi.Inode()) + originalXattrs := map[string][]byte{ + s3Etag: []byte("original-etag"), + s3Tags: []byte("owner=posix"), + s3Meta: []byte(`{"x-amz-meta-owner":"posix"}`), + } + for name, value := range originalXattrs { + if eno = jfs.SetXattr(mctx, directoryPath, name, value, 0); eno != 0 { + t.Fatalf("set original xattr %s: %s", name, eno) + } + } + assertHeadObject(t, gateway, bucket, "prefix/", false) + + _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{IfNoneMatch: true}) + if err != nil { + t.Fatalf("promote implicit directory: %v", err) + } + after, eno := jfs.Stat(mctx, directoryPath) + if eno != 0 { + t.Fatalf("stat rejected implicit directory: %s", eno) + } + if after.Inode() != fi.Inode() || !isExplicitDirectoryMarker(after.Attr()) { + t.Fatalf("promotion did not preserve inode and publish marker: %+v", after) + } + for name := range originalXattrs { + if _, eno := jfs.GetXattr(mctx, directoryPath, name); eno != meta.ENOATTR { + t.Fatalf("old xattr %s remains: %s", name, eno) + } + } + assertHeadObject(t, gateway, bucket, "prefix/", true) + _, err = gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + if got := readGatewayObject(t, gateway, bucket, "prefix/child"); !bytes.Equal(got, []byte("child")) { + t.Fatalf("child object changed: got %q, want %q", got, "child") + } + }) + + t.Run("non-empty directory PUT does not create a marker", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + assertHeadObject(t, gateway, bucket, "prefix/", false) + + _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, []byte("not-empty")), minio.ObjectOptions{IfNoneMatch: true}) + var existsAsDirectory minio.ObjectExistsAsDirectory + if !errors.As(err, &existsAsDirectory) { + t.Fatalf("expected ObjectExistsAsDirectory, got %T: %v", err, err) + } + assertHeadObject(t, gateway, bucket, "prefix/", false) + }) + + t.Run("sub-millisecond atime uses the same marker semantics", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if eno := jfs.MkdirAll(mctx, "/prefix", 0777, 022); eno != 0 { + t.Fatalf("create directory: %s", eno) + } + fi, eno := jfs.Stat(mctx, "/prefix") + if eno != 0 { + t.Fatalf("stat directory: %s", eno) + } + attr := meta.Attr{Atime: 0, Atimensec: 1} + if eno = jfs.Meta().SetAttr(mctx, fi.Inode(), meta.SetAttrAtime, 0, &attr); eno != 0 { + t.Fatalf("set directory atime: %s", eno) + } + jfs.InvalidateAttr(fi.Inode()) + + assertHeadObject(t, gateway, bucket, "prefix/", true) + _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + }) + + t.Run("unconditional PUT still promotes an implicit directory", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + if _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{}); err != nil { + t.Fatalf("unconditionally create marker for implicit directory: %s", err) + } + assertHeadObject(t, gateway, bucket, "prefix/", true) + if got := readGatewayObject(t, gateway, bucket, "prefix/child"); !bytes.Equal(got, []byte("child")) { + t.Fatalf("child object changed: got %q, want %q", got, "child") + } + }) + + t.Run("concurrent POSIX directory creation has a safe outcome", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + const iterations = 128 + for i := 0; i < iterations; i++ { + object := fmt.Sprintf("prefix-%03d/", i) + directoryPath := gateway.path(bucket, object) + reader := newPutObjectReader(t, nil) + start := make(chan struct{}) + markerResult := make(chan error, 1) + directoryResult := make(chan error, 1) + + go func() { + <-start + _, err := gateway.PutObject(ctx, bucket, object, reader, minio.ObjectOptions{IfNoneMatch: true}) + markerResult <- err + }() + go func() { + <-start + directoryResult <- gateway.mkdirAll(ctx, directoryPath) + }() + close(start) + + if err := <-directoryResult; err != nil { + t.Fatalf("create POSIX directory %s: %s", directoryPath, err) + } + markerErr := <-markerResult + if markerErr == nil { + assertHeadObject(t, gateway, bucket, object, true) + continue + } + t.Fatalf("conditional marker creation: %v", markerErr) + } + }) + + t.Run("head dir treats an implicit directory as existing", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{HeadDir: true}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + _, err := gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + }) +} + +func TestCopyObjectDirectoryMarker(t *testing.T) { + for _, destination := range []string{"new", "implicit", "explicit", "self"} { + t.Run(destination, func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, nil), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + var inode meta.Ino + if destination != "new" { + if _, err = gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child: %s", err) + } + if destination != "implicit" { + if _, err = gateway.PutObject(ctx, bucket, "prefix/", newPutObjectReader(t, nil), minio.ObjectOptions{}); err != nil { + t.Fatalf("create marker: %s", err) + } + } + fi, eno := jfs.Stat(mctx, gateway.path(bucket, "prefix/")) + if eno != 0 { + t.Fatalf("stat directory: %s", eno) + } + inode = fi.Inode() + } + + for _, owner := range []string{"first", "replacement", ""} { + source := "source" + if destination == "self" { + source = "prefix/" + srcInfo, err = gateway.GetObjectInfo(ctx, bucket, source, minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get self-copy source: %s", err) + } + } + tags := "" + srcInfo.UserDefined = map[string]string{} + if owner != "" { + tags = "owner=" + owner + srcInfo.UserDefined["x-amz-meta-owner"] = owner + srcInfo.UserDefined[xhttp.AmzObjectTagging] = tags + } + + copyInfo, err := gateway.CopyObject(ctx, bucket, source, bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{}) + if err != nil { + t.Fatalf("copy marker with owner %q: %s", owner, err) + } + + info, err := gateway.GetObjectInfo(ctx, bucket, "prefix/", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get copied marker: %s", err) + } + if !info.IsDir || info.Size != 0 || info.ETag != srcInfo.ETag || copyInfo.ETag != info.ETag || + info.UserDefined["x-amz-meta-owner"] != owner || info.UserTags != tags { + t.Fatalf("unexpected copied marker with owner %q: %+v", owner, info) + } + fi, eno := jfs.Stat(mctx, gateway.path(bucket, "prefix/")) + if eno != 0 { + t.Fatalf("stat copied marker: %s", eno) + } + if !isExplicitDirectoryMarker(fi.Attr()) || inode != 0 && fi.Inode() != inode { + t.Fatalf("copy did not preserve directory inode and publish marker: %+v", fi) + } + inode = fi.Inode() + if destination != "new" { + if got := readGatewayObject(t, gateway, bucket, "prefix/child"); !bytes.Equal(got, []byte("child")) { + t.Fatalf("child changed: %q", got) + } + } + } + }) + } +} + +func TestCopyObjectRejectsNonEmptyDirectoryMarker(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, []byte("source")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{}) + + var existsAsDirectory minio.ObjectExistsAsDirectory + if !errors.As(err, &existsAsDirectory) { + t.Fatalf("expected ObjectExistsAsDirectory, got %T: %v", err, err) + } + if _, eno := jfs.Stat(mctx, gateway.path(bucket, "prefix/")); !fs.IsNotExist(eno) { + t.Fatalf("rejected copy created a destination: %s", eno) + } +} + +func TestCopyObjectIfNoneMatch(t *testing.T) { + t.Run("existing destination remains unchanged", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, []byte("source")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + if _, err := gateway.PutObject(ctx, bucket, "destination", newPutObjectReader(t, []byte("original")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create destination: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "destination", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + if got := readGatewayObject(t, gateway, bucket, "destination"); !bytes.Equal(got, []byte("original")) { + t.Fatalf("destination changed: got %q, want %q", got, "original") + } + }) + + t.Run("new destination is created", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, []byte("source")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + if _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "destination", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}); err != nil { + t.Fatalf("conditionally copy to new destination: %s", err) + } + if got := readGatewayObject(t, gateway, bucket, "destination"); !bytes.Equal(got, []byte("source")) { + t.Fatalf("copied data: got %q, want %q", got, "source") + } + }) + + t.Run("copying an object onto itself fails", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, []byte("source")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "source", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + }) + + t.Run("zero-byte source creates a new marker with complete attributes", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + sourceOpts := minio.ObjectOptions{UserDefined: map[string]string{"x-amz-meta-owner": "source"}} + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, nil), sourceOpts); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + srcInfo.UserDefined["x-amz-meta-owner"] = "final" + srcInfo.UserDefined[xhttp.AmzObjectTagging] = "project=juicefs" + + copyInfo, err := gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + if err != nil { + t.Fatalf("conditionally copy to new marker: %s", err) + } + assertHeadObject(t, gateway, bucket, "prefix/", true) + if copyInfo.ETag != srcInfo.ETag { + t.Fatalf("copy ETag: got %q, want %q", copyInfo.ETag, srcInfo.ETag) + } + dstInfo, err := gateway.GetObjectInfo(ctx, bucket, "prefix/", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get destination info: %s", err) + } + if dstInfo.ETag != copyInfo.ETag { + t.Fatalf("destination HEAD ETag: got %q, want %q", dstInfo.ETag, copyInfo.ETag) + } + listInfo, err := gateway.ListObjects(ctx, bucket, "prefix/", "", "", 100) + if err != nil { + t.Fatalf("list destination marker: %s", err) + } + var listedMarker *minio.ObjectInfo + for i := range listInfo.Objects { + if listInfo.Objects[i].Name == "prefix/" { + listedMarker = &listInfo.Objects[i] + break + } + } + if listedMarker == nil { + t.Fatalf("destination marker missing from LIST: %#v", listInfo.Objects) + } + if listedMarker.ETag != copyInfo.ETag { + t.Fatalf("destination LIST ETag: got %q, want %q", listedMarker.ETag, copyInfo.ETag) + } + if got := dstInfo.UserDefined["x-amz-meta-owner"]; got != "final" { + t.Fatalf("destination metadata: got %q, want %q", got, "final") + } + if dstInfo.UserTags != "project=juicefs" { + t.Fatalf("destination tags: got %q, want %q", dstInfo.UserTags, "project=juicefs") + } + storedEtag, eno := jfs.GetXattr(mctx, "/prefix", s3Etag) + if eno != 0 { + t.Fatalf("get destination ETag: %s", eno) + } + if string(storedEtag) != srcInfo.ETag { + t.Fatalf("stored ETag: got %q, want %q", storedEtag, srcInfo.ETag) + } + fi, eno := jfs.Stat(mctx, "/prefix") + if eno != 0 { + t.Fatalf("stat destination marker: %s", eno) + } + if !isExplicitDirectoryMarker(fi.Attr()) { + t.Fatalf("destination was not published as an explicit marker: atime=%d.%09d", fi.Attr().Atime, fi.Attr().Atimensec) + } + }) + + t.Run("zero-byte source promotes an implicit directory atomically", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + sourceOpts := minio.ObjectOptions{UserDefined: map[string]string{ + "x-amz-meta-owner": "source", + xhttp.AmzObjectTagging: "owner=source", + }} + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, nil), sourceOpts); err != nil { + t.Fatalf("create source: %s", err) + } + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + directoryPath := gateway.path(bucket, "prefix") + fi, eno := jfs.Stat(mctx, directoryPath) + if eno != 0 { + t.Fatalf("stat implicit directory: %s", eno) + } + attr := meta.Attr{Atime: 321, Atimensec: 654000000} + if eno = jfs.Meta().SetAttr(mctx, fi.Inode(), meta.SetAttrAtime, 0, &attr); eno != 0 { + t.Fatalf("set implicit directory atime: %s", eno) + } + jfs.InvalidateAttr(fi.Inode()) + originalXattrs := map[string][]byte{ + s3Etag: []byte("posix-etag"), + s3Tags: []byte("owner=posix"), + s3Meta: []byte(`{"x-amz-meta-owner":"posix"}`), + } + for name, value := range originalXattrs { + if eno = jfs.SetXattr(mctx, directoryPath, name, value, 0); eno != 0 { + t.Fatalf("set original xattr %s: %s", name, eno) + } + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + srcInfo.UserDefined[xhttp.AmzObjectTagging] = srcInfo.UserTags + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + if err != nil { + t.Fatalf("promote directory by copy: %v", err) + } + after, eno := jfs.Stat(mctx, directoryPath) + if eno != 0 { + t.Fatalf("stat rejected implicit destination: %s", eno) + } + if after.Inode() != fi.Inode() || !isExplicitDirectoryMarker(after.Attr()) { + t.Fatalf("copy did not preserve inode and publish marker: %+v", after) + } + info, err := gateway.GetObjectInfo(ctx, bucket, "prefix/", minio.ObjectOptions{}) + if err != nil || info.ETag != srcInfo.ETag || info.UserTags != "owner=source" || info.UserDefined["x-amz-meta-owner"] != "source" { + t.Fatalf("copied marker attributes: %+v, %v", info, err) + } + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + assertHeadObject(t, gateway, bucket, "prefix/", true) + if got := readGatewayObject(t, gateway, bucket, "prefix/child"); !bytes.Equal(got, []byte("child")) { + t.Fatalf("child object changed: got %q, want %q", got, "child") + } + }) + + t.Run("non-empty source does not create a directory marker", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, []byte("source")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + var existsAsDirectory minio.ObjectExistsAsDirectory + if !errors.As(err, &existsAsDirectory) { + t.Fatalf("expected ObjectExistsAsDirectory, got %T: %v", err, err) + } + assertHeadObject(t, gateway, bucket, "prefix/", false) + }) + + t.Run("head dir treats an implicit destination as existing", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{HeadDir: true}) + ctx := context.Background() + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, nil), minio.ObjectOptions{}); err != nil { + t.Fatalf("create source: %s", err) + } + if _, err := gateway.PutObject(ctx, bucket, "prefix/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child object: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + _, err = gateway.CopyObject(ctx, bucket, "source", bucket, "prefix/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + fi, eno := jfs.Stat(mctx, "/prefix") + if eno != 0 { + t.Fatalf("stat implicit directory: %s", eno) + } + if isExplicitDirectoryMarker(fi.Attr()) { + t.Fatal("failed conditional copy changed the implicit directory into a marker") + } + }) + + t.Run("unconditional PUT and conditional directory copy preserve attributes", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + sourceOpts := minio.ObjectOptions{UserDefined: map[string]string{ + "x-amz-meta-owner": "copy", + xhttp.AmzObjectTagging: "winner=copy", + }} + if _, err := gateway.PutObject(ctx, bucket, "source", newPutObjectReader(t, nil), sourceOpts); err != nil { + t.Fatalf("create source: %s", err) + } + srcInfo, err := gateway.GetObjectInfo(ctx, bucket, "source", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("get source info: %s", err) + } + + const iterations = 128 + for i := 0; i < iterations; i++ { + object := fmt.Sprintf("concurrent-prefix-%03d", i) + implicitDirectory := i%2 == 0 + if implicitDirectory { + if _, err = gateway.PutObject(ctx, bucket, object+"/child", newPutObjectReader(t, []byte("child")), minio.ObjectOptions{}); err != nil { + t.Fatalf("create child for %s: %s", object, err) + } + } + + start := make(chan struct{}) + putResult := make(chan error, 1) + copyResult := make(chan error, 1) + putReader := newPutObjectReader(t, nil) + go func() { + <-start + _, putErr := gateway.PutObject(ctx, bucket, object+"/", putReader, minio.ObjectOptions{}) + putResult <- putErr + }() + go func() { + <-start + _, copyErr := gateway.CopyObject(ctx, bucket, "source", bucket, object+"/", srcInfo, + minio.ObjectOptions{}, minio.ObjectOptions{IfNoneMatch: true}) + copyResult <- copyErr + }() + close(start) + + if putErr := <-putResult; putErr != nil { + t.Fatalf("unconditional marker PUT for %s: %s", object, putErr) + } + copyErr := <-copyResult + var preconditionFailed minio.PreConditionFailed + if copyErr != nil && !errors.As(copyErr, &preconditionFailed) { + t.Fatalf("conditional marker Copy for %s: %T: %v", object, copyErr, copyErr) + } + + assertHeadObject(t, gateway, bucket, object+"/", true) + info, infoErr := gateway.GetObjectInfo(ctx, bucket, object+"/", minio.ObjectOptions{}) + if infoErr != nil { + t.Fatalf("get final marker %s: %s", object, infoErr) + } + if info.ETag != "" || info.UserTags != "" || info.UserDefined["x-amz-meta-owner"] != "" { + t.Fatalf("copy attributes leaked into PUT marker %s: ETag=%q tags=%q metadata=%q", object, info.ETag, info.UserTags, info.UserDefined) + } + for _, name := range []string{s3Etag, s3Tags, s3Meta} { + if value, eno := jfs.GetXattr(mctx, gateway.path(bucket, object), name); eno != meta.ENOATTR { + t.Fatalf("copy xattr %s leaked into PUT marker %s: value=%q, errno=%s", name, object, value, eno) + } + } + } + }) + + t.Run("temporary marker inode is removed after publish failure", func(t *testing.T) { + gateway, jfs, bucket := newTestGateway(t, Config{KeepEtag: true, ObjTag: true, ObjMeta: true}) + ctx := context.Background() + var countTemporaryObjects func(string) int + countTemporaryObjects = func(directory string) int { + t.Helper() + f, eno := jfs.Open(mctx, directory, 0) + if fs.IsNotExist(eno) { + return 0 + } + if eno != 0 { + t.Fatalf("open temporary directory %s: %s", directory, eno) + } + entries, eno := f.ReaddirPlus(mctx, 0) + if closeErr := f.Close(mctx); closeErr != 0 { + t.Fatalf("close temporary directory %s: %s", directory, closeErr) + } + if eno != 0 { + t.Fatalf("read temporary directory %s: %s", directory, eno) + } + count := 0 + for _, entry := range entries { + if entry.Attr.Typ == meta.TypeDirectory && len(entry.Name) == subDirPrefix { + count += countTemporaryObjects(directory + sep + string(entry.Name)) + continue + } + count++ + } + return count + } + tmpRoot := gateway.tpath(bucket, "tmp") + invalidXattr := objectXattr{name: "", value: []byte("invalid")} + before := countTemporaryObjects(tmpRoot) + if err := gateway.putDirectoryObject(ctx, bucket, "/new-marker", true, []objectXattr{invalidXattr}); err == nil { + t.Fatal("expected new marker attribute failure") + } + if _, eno := jfs.Stat(mctx, "/new-marker"); !fs.IsNotExist(eno) { + t.Fatalf("failed marker became visible: %s", eno) + } + if after := countTemporaryObjects(tmpRoot); after != before { + t.Fatalf("attribute failure leaked temporary marker: before=%d after=%d", before, after) + } + + if _, err := gateway.PutObject(ctx, bucket, "existing-marker/", newPutObjectReader(t, nil), minio.ObjectOptions{}); err != nil { + t.Fatalf("create existing marker: %s", err) + } + before = countTemporaryObjects(tmpRoot) + err := gateway.publishNewDirectoryObject(ctx, bucket, gateway.path(bucket, "existing-marker"), nil) + if !errors.Is(err, syscall.EEXIST) { + t.Fatalf("expected marker publish collision, got %T: %v", err, err) + } + if after := countTemporaryObjects(tmpRoot); after != before { + t.Fatalf("publish collision leaked temporary marker: before=%d after=%d", before, after) + } + + before = countTemporaryObjects(tmpRoot) + if err = gateway.putDirectoryObject(ctx, bucket, gateway.path(bucket, "successful-marker"), true, nil); err != nil { + t.Fatalf("publish new marker: %s", err) + } + if after := countTemporaryObjects(tmpRoot); after != before { + t.Fatalf("successful publish left temporary marker: before=%d after=%d", before, after) + } + }) +} + +func TestCompleteMultipartUploadIfNoneMatch(t *testing.T) { + t.Run("existing object remains unchanged and a new object is created", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + original := []byte("original") + if _, err := gateway.PutObject(ctx, bucket, "object", newPutObjectReader(t, original), minio.ObjectOptions{}); err != nil { + t.Fatalf("create original object: %s", err) + } + + uploadID, err := gateway.NewMultipartUpload(ctx, bucket, "object", minio.ObjectOptions{}) + if err != nil { + t.Fatalf("create multipart upload: %s", err) + } + part, err := gateway.PutObjectPart(ctx, bucket, "object", uploadID, 1, newPutObjectReader(t, []byte("replacement")), minio.ObjectOptions{}) + if err != nil { + t.Fatalf("put multipart part: %s", err) + } + _, err = gateway.CompleteMultipartUpload(ctx, bucket, "object", uploadID, + []minio.CompletePart{{PartNumber: 1, ETag: part.ETag}}, minio.ObjectOptions{IfNoneMatch: true}) + assertPreconditionFailed(t, err) + if got := readGatewayObject(t, gateway, bucket, "object"); !bytes.Equal(got, original) { + t.Fatalf("existing object changed: got %q, want %q", got, original) + } + + newObject := "new-object" + uploadID, err = gateway.NewMultipartUpload(ctx, bucket, newObject, minio.ObjectOptions{}) + if err != nil { + t.Fatalf("create multipart upload for new object: %s", err) + } + part, err = gateway.PutObjectPart(ctx, bucket, newObject, uploadID, 1, + newPutObjectReader(t, []byte("created")), minio.ObjectOptions{}) + if err != nil { + t.Fatalf("put multipart part for new object: %s", err) + } + _, err = gateway.CompleteMultipartUpload(ctx, bucket, newObject, uploadID, + []minio.CompletePart{{PartNumber: 1, ETag: part.ETag}}, minio.ObjectOptions{IfNoneMatch: true}) + if err != nil { + t.Fatalf("conditionally complete new object: %s", err) + } + if got := readGatewayObject(t, gateway, bucket, newObject); !bytes.Equal(got, []byte("created")) { + t.Fatalf("new multipart object: got %q, want %q", got, "created") + } + }) + + t.Run("concurrent completions have one winner", func(t *testing.T) { + gateway, _, bucket := newTestGateway(t, Config{}) + ctx := context.Background() + const writers = 8 + const object = "race" + type attempt struct { + uploadID string + part minio.CompletePart + payload []byte + } + attempts := make([]attempt, writers) + for i := range attempts { + payload := []byte(fmt.Sprintf("multipart-writer-%02d", i)) + uploadID, err := gateway.NewMultipartUpload(ctx, bucket, object, minio.ObjectOptions{}) + if err != nil { + t.Fatalf("create multipart upload %d: %s", i, err) + } + part, err := gateway.PutObjectPart(ctx, bucket, object, uploadID, 1, newPutObjectReader(t, payload), minio.ObjectOptions{}) + if err != nil { + t.Fatalf("put multipart part %d: %s", i, err) + } + attempts[i] = attempt{ + uploadID: uploadID, + part: minio.CompletePart{PartNumber: 1, ETag: part.ETag}, + payload: payload, + } + } + + type result struct { + index int + err error + } + start := make(chan struct{}) + results := make(chan result, writers) + var waitGroup sync.WaitGroup + for i := range attempts { + waitGroup.Add(1) + go func(index int) { + defer waitGroup.Done() + <-start + attempt := attempts[index] + _, err := gateway.CompleteMultipartUpload(ctx, bucket, object, attempt.uploadID, + []minio.CompletePart{attempt.part}, minio.ObjectOptions{IfNoneMatch: true}) + results <- result{index: index, err: err} + }(i) + } + close(start) + waitGroup.Wait() + close(results) + + succeeded := 0 + failed := 0 + winningIndex := -1 + failedIndex := -1 + for result := range results { + if result.err == nil { + succeeded++ + winningIndex = result.index + continue + } + var preconditionFailed minio.PreConditionFailed + if errors.As(result.err, &preconditionFailed) { + failed++ + failedIndex = result.index + continue + } + t.Fatalf("unexpected concurrent multipart error: %T: %v", result.err, result.err) + } + if succeeded != 1 || failed != writers-1 { + t.Fatalf("conditional multipart results: succeeded=%d failed=%d", succeeded, failed) + } + if got := readGatewayObject(t, gateway, bucket, object); !bytes.Equal(got, attempts[winningIndex].payload) { + t.Fatalf("stored payload %q does not match successful upload %q", got, attempts[winningIndex].payload) + } + + if _, err := gateway.DeleteObject(ctx, bucket, object, minio.ObjectOptions{}); err != nil { + t.Fatalf("delete winning object: %s", err) + } + failedAttempt := attempts[failedIndex] + if _, err := gateway.CompleteMultipartUpload(ctx, bucket, object, failedAttempt.uploadID, + []minio.CompletePart{failedAttempt.part}, minio.ObjectOptions{IfNoneMatch: true}); err != nil { + t.Fatalf("retry losing multipart upload: %s", err) + } + if got := readGatewayObject(t, gateway, bucket, object); !bytes.Equal(got, failedAttempt.payload) { + t.Fatalf("retried multipart payload: got %q, want %q", got, failedAttempt.payload) + } + }) +} + func assertHeadObject(t *testing.T, jfsObj *jfsObjects, bucket, object string, wantFound bool) { t.Helper() _, err := jfsObj.GetObjectInfo(context.Background(), bucket, object, minio.ObjectOptions{}) diff --git a/pkg/meta/base.go b/pkg/meta/base.go index bb1f26ee7869..9746f737ef59 100644 --- a/pkg/meta/base.go +++ b/pkg/meta/base.go @@ -116,6 +116,7 @@ type engine interface { doGetAttr(ctx Context, inode Ino, attr *Attr) syscall.Errno doSetAttr(ctx Context, inode Ino, set uint16, sugidclearmode uint8, attr *Attr, oldAttr *Attr) syscall.Errno + doSetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno doLookup(ctx Context, parent Ino, name string, inode *Ino, attr *Attr) syscall.Errno doMknod(ctx Context, parent Ino, name string, _type uint8, mode, cumask uint16, path string, inode *Ino, attr *Attr) syscall.Errno doLink(ctx Context, inode, parent Ino, name string, attr *Attr) syscall.Errno @@ -1529,6 +1530,75 @@ func (m *baseMeta) SetAttr(ctx Context, inode Ino, set uint16, sugidclearmode ui return err } +func (m *baseMeta) SetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno { + if m.conf.ReadOnly { + return syscall.EROFS + } + if name == "" || name == "." || name == ".." || strings.ContainsAny(name, "/\x00") || attr == nil { + return syscall.EINVAL + } + if len(name) > MaxName { + return syscall.ENAMETOOLONG + } + for key := range xattrs { + if key == "" || strings.ContainsRune(key, 0) { + return syscall.EINVAL + } + } + if ctx.Canceled() { + return syscall.EINTR + } + defer m.timeit("SetDirMarker", time.Now()) + parent, inode = m.checkRoot(parent), m.checkRoot(inode) + if m.conf.CaseInsensi { + if entry := m.resolveCase(ctx, parent, name); entry != nil { + name = string(entry.Name) + } + } + st := m.en.doSetDirMarker(ctx, parent, name, inode, exclusive, xattrs, attr) + if st == 0 { + m.of.InvalidateChunk(inode, invalidateAttrOnly) + m.of.Update(inode, attr) + } + return st +} + +// prepareDirMarker checks and prepares the attribute update inside the backend +// transaction, before any xattrs are modified. +func (m *baseMeta) prepareDirMarker(ctx Context, parent, inode Ino, parentAttr, attr *Attr, exclusive bool, now time.Time) syscall.Errno { + if parentAttr.Typ != TypeDirectory || attr.Typ != TypeDirectory { + return syscall.ENOTDIR + } + if parent.IsTrash() || attr.Parent > TrashInode || attr.Flags&(FlagImmutable|FlagAppend) != 0 { + return syscall.EPERM + } + if st := m.Access(ctx, parent, MODE_MASK_X, parentAttr); st != 0 { + return st + } + if exclusive && attr.Atime*1000+int64(attr.Atimensec/1e6) == 0 { + return syscall.EEXIST + } + if st := m.Access(ctx, inode, MODE_MASK_W, attr); st != 0 { + return st + } + updated, st := m.mergeAttr(ctx, inode, SetAttrAtime, attr, &Attr{}, now, nil) + if st != 0 { + return st + } + if updated != nil { + *attr = *updated + } + attr.Ctime, attr.Ctimensec = now.Unix(), uint32(now.Nanosecond()) + return 0 +} + +// One log record preserves the transaction boundary, including on KV backends +// whose log key is the transaction timestamp. JSON null means remove an xattr. +func dirMarkerLog(inode Ino, attr *Attr, xattrs map[string][]byte) string { + values, _ := json.Marshal(xattrs) // map[string][]byte is always encodable + return fmt.Sprintf("SETDIRMARKER(%d,%d,%d,%s)", inode, attr.Ctime, attr.Ctimensec, logEncode(values)) +} + func (m *baseMeta) nextInode() (Ino, error) { m.freeMu.Lock() defer m.freeMu.Unlock() diff --git a/pkg/meta/base_test.go b/pkg/meta/base_test.go index 972b2f0f3266..a87aeb50fefe 100644 --- a/pkg/meta/base_test.go +++ b/pkg/meta/base_test.go @@ -183,6 +183,7 @@ func testMeta(t *testing.T, m Meta) { testConcurrentWrite(t, m) testRace(t, m) testXattr(t, m) + t.Run("DirMarker", func(t *testing.T) { testDirMarker(t, m) }) testCompaction(t, m, false) time.Sleep(time.Second) testCompaction(t, m, true) diff --git a/pkg/meta/dir_marker_test.go b/pkg/meta/dir_marker_test.go new file mode 100644 index 000000000000..188998884e67 --- /dev/null +++ b/pkg/meta/dir_marker_test.go @@ -0,0 +1,336 @@ +/* + * JuiceFS, Copyright 2020 Juicedata, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// +//mutate:disable +//nolint:errcheck +package meta + +import ( + "encoding/json" + "errors" + "fmt" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDirMarker(t *testing.T) { + cases := map[string]string{ + "memkv": "memkv://", + "sqlite": "sqlite3://" + filepath.Join(t.TempDir(), "marker.db"), + "badger": "badger://" + t.TempDir(), + } + for _, engine := range []string{"redis", "valkey", "keydb", "mysql", "mariadb", "tidb", "oceanbase", "postgres", "cockroach", "etcd", "tikv", "fdb"} { + if uri := os.Getenv("JFS_MARKER_TEST_" + strings.ToUpper(engine)); uri != "" { + cases[engine] = uri + } + } + for name, uri := range cases { + t.Run(name, func(t *testing.T) { + m := NewClient(uri, testConfig()) + require.NoError(t, m.Reset()) + format := testFormat() + format.ChangeLog = name != "etcd" + require.NoError(t, m.Init(format, true)) + t.Cleanup(func() { require.NoError(t, m.Shutdown()) }) + testDirMarker(t, m) + if name != "memkv" && name != "badger" { + t.Run("multiple clients", func(t *testing.T) { + other := NewClient(uri, testConfig()) + _, err := other.Load(false) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, other.Shutdown()) }) + testDirMarkerWriters(t, []Meta{m, other}) + }) + } + if format.ChangeLog { + t.Run("changelog", func(t *testing.T) { testDirMarkerChangelog(t, m) }) + } + if r, ok := m.(*redisMeta); ok { + t.Run("invalid xattr storage", func(t *testing.T) { testDirMarkerRedisFailure(t, r) }) + } + }) + } +} + +func testDirMarker(t *testing.T, m Meta) { + ctx := Background() + var seq int + newDirectory := func(t *testing.T) (string, Ino, Attr) { + t.Helper() + seq++ + name := fmt.Sprintf("dir-marker-%d", seq) + var inode Ino + var attr Attr + require.Zero(t, m.Mkdir(ctx, RootInode, name, 0755, 0, 0, &inode, &attr)) + return name, inode, attr + } + getXattr := func(t *testing.T, inode Ino, key string) []byte { + t.Helper() + var value []byte + require.Zero(t, m.GetXattr(ctx, inode, key, &value)) + return value + } + t.Run("promotion preserves inode children and unrelated attributes", func(t *testing.T) { + name, inode, before := newDirectory(t) + var child Ino + require.Zero(t, m.Mknod(ctx, inode, "child", TypeFile, 0644, 0, 0, "", &child, nil)) + require.Zero(t, m.SetXattr(ctx, inode, "user.old", []byte("old"), 0)) + require.Zero(t, m.SetXattr(ctx, inode, "user.posix", []byte("keep"), 0)) + var result Attr + updates := map[string][]byte{"user.old": nil, "user.new": []byte("new"), "user.missing": nil} + require.Zero(t, m.SetDirMarker(ctx, RootInode, name, inode, true, updates, &result)) + require.Zero(t, result.Atime) + require.Zero(t, result.Atimensec) + require.Equal(t, before.Uid, result.Uid) + require.Equal(t, before.Gid, result.Gid) + require.Equal(t, before.Mode, result.Mode) + require.Equal(t, before.Parent, result.Parent) + require.Equal(t, []byte("new"), getXattr(t, inode, "user.new")) + require.Equal(t, []byte("keep"), getXattr(t, inode, "user.posix")) + var value []byte + require.Equal(t, ENOATTR, m.GetXattr(ctx, inode, "user.old", &value)) + var found Ino + require.Zero(t, m.Lookup(ctx, inode, "child", &found, &result, true)) + require.Equal(t, child, found) + require.Equal(t, syscall.EEXIST, m.SetDirMarker(ctx, RootInode, name, inode, true, map[string][]byte{"user.new": []byte("loser")}, &result)) + require.Equal(t, []byte("new"), getXattr(t, inode, "user.new")) + require.Zero(t, m.SetDirMarker(ctx, RootInode, name, inode, false, map[string][]byte{"user.new": []byte("overwrite")}, &result)) + require.Equal(t, []byte("overwrite"), getXattr(t, inode, "user.new")) + }) + t.Run("rejected updates leave all state unchanged", func(t *testing.T) { + for _, scenario := range []string{"invalid xattr", "permission", "existing marker permission", "immutable", "append only", "read only", "canceled", "existing marker"} { + t.Run(scenario, func(t *testing.T) { + name, inode, _ := newDirectory(t) + require.Zero(t, m.SetXattr(ctx, inode, "user.keep", []byte("old"), 0)) + updates := map[string][]byte{"user.keep": []byte("new"), "user.new": []byte("new")} + requestCtx := ctx + exclusive := true + expected := syscall.EINVAL + switch scenario { + case "invalid xattr": + updates[""] = []byte("invalid") + case "permission": + requestCtx = NewContext(1, 1000, []uint32{1000}) + expected = syscall.EACCES + case "existing marker permission": + a := Attr{} + require.Zero(t, m.SetAttr(ctx, inode, SetAttrAtime, 0, &a)) + requestCtx = NewContext(1, 1000, []uint32{1000}) + exclusive = false + expected = syscall.EACCES + case "immutable", "append only": + flag := uint8(FlagImmutable) + if scenario == "append only" { + flag = FlagAppend + } + a := Attr{Flags: flag} + require.Zero(t, m.SetAttr(ctx, inode, SetAttrFlag, 0, &a)) + expected = syscall.EPERM + case "read only": + m.getBase().conf.ReadOnly = true + defer func() { m.getBase().conf.ReadOnly = false }() + expected = syscall.EROFS + case "canceled": + requestCtx = NewContext(1, 0, []uint32{0}) + requestCtx.Cancel() + expected = syscall.EINTR + case "existing marker": + a := Attr{} + require.Zero(t, m.SetAttr(ctx, inode, SetAttrAtime, 0, &a)) + expected = syscall.EEXIST + } + var before, after, result Attr + require.Zero(t, m.getBase().en.doGetAttr(ctx, inode, &before)) + require.Equal(t, expected, m.SetDirMarker(requestCtx, RootInode, name, inode, exclusive, updates, &result)) + require.Zero(t, m.getBase().en.doGetAttr(ctx, inode, &after)) + require.Equal(t, before, after) + require.Equal(t, []byte("old"), getXattr(t, inode, "user.keep")) + var value []byte + require.Equal(t, ENOATTR, m.GetXattr(ctx, inode, "user.new", &value)) + }) + } + }) + t.Run("renamed and replaced paths are not modified", func(t *testing.T) { + name, inode, before := newDirectory(t) + require.Zero(t, m.Rename(ctx, RootInode, name, RootInode, name+"-moved", 0, nil, nil)) + var result Attr + updates := map[string][]byte{"user.new": []byte("new")} + require.Equal(t, syscall.ENOENT, m.SetDirMarker(ctx, RootInode, name, inode, true, updates, &result)) + var replacement Ino + require.Zero(t, m.Mkdir(ctx, RootInode, name, 0755, 0, 0, &replacement, nil)) + require.Equal(t, syscall.EAGAIN, m.SetDirMarker(ctx, RootInode, name, inode, true, updates, &result)) + require.Zero(t, m.getBase().en.doGetAttr(ctx, inode, &result)) + require.Equal(t, before.Atime, result.Atime) + require.Equal(t, before.Atimensec, result.Atimensec) + var value []byte + require.Equal(t, ENOATTR, m.GetXattr(ctx, inode, "user.new", &value)) + require.Equal(t, ENOATTR, m.GetXattr(ctx, replacement, "user.new", &value)) + }) + t.Run("concurrent conditional writers have one complete winner", func(t *testing.T) { testDirMarkerWriters(t, []Meta{m}) }) + + t.Run("rename race never promotes a replacement directory", func(t *testing.T) { + for i := 0; i < 32; i++ { + name, inode, _ := newDirectory(t) + start := make(chan struct{}) + result := make(chan syscall.Errno, 1) + go func() { + <-start + var attr Attr + result <- m.SetDirMarker(ctx, RootInode, name, inode, true, map[string][]byte{"user.winner": []byte("original")}, &attr) + }() + close(start) + require.Zero(t, m.Rename(ctx, RootInode, name, RootInode, name+"-moved", 0, nil, nil)) + var replacement Ino + require.Zero(t, m.Mkdir(ctx, RootInode, name, 0755, 0, 0, &replacement, nil)) + st := <-result + require.Contains(t, []syscall.Errno{0, syscall.ENOENT, syscall.EAGAIN}, st) + var attr Attr + require.Zero(t, m.getBase().en.doGetAttr(ctx, replacement, &attr)) + require.NotZero(t, attr.Atime) + var value []byte + require.Equal(t, ENOATTR, m.GetXattr(ctx, replacement, "user.winner", &value)) + } + }) + t.Run("case insensitive names", func(t *testing.T) { + name, inode, _ := newDirectory(t) + old := m.getBase().conf.CaseInsensi + m.getBase().conf.CaseInsensi = true + defer func() { m.getBase().conf.CaseInsensi = old }() + var attr Attr + require.Zero(t, m.SetDirMarker(ctx, RootInode, strings.ToUpper(name), inode, true, nil, &attr)) + }) +} + +func TestDirMarkerSQLiteRollback(t *testing.T) { + client, err := newSQLMeta("sqlite3", filepath.Join(t.TempDir(), "rollback.db"), testConfig()) + require.NoError(t, err) + m := client.(*dbMeta) + require.NoError(t, m.Init(testFormat(), true)) + t.Cleanup(func() { require.NoError(t, m.Shutdown()) }) + ctx := Background() + var inode Ino + var before, after Attr + require.Zero(t, m.Mkdir(ctx, RootInode, "dir", 0755, 0, 0, &inode, &before)) + require.Zero(t, m.SetXattr(ctx, inode, "user.old", []byte("original"), 0)) + // Abort the final inode write, after the xattr changes have executed. + _, err = m.db.Exec(`CREATE TRIGGER fail_marker BEFORE UPDATE OF atime ON jfs_node BEGIN SELECT RAISE(ABORT, 'injected marker failure'); END`) + require.NoError(t, err) + require.NotZero(t, m.SetDirMarker(ctx, RootInode, "dir", inode, true, map[string][]byte{"user.old": nil, "user.new": []byte("new")}, &after)) + require.Zero(t, m.doGetAttr(ctx, inode, &after)) + require.Equal(t, before, after) + var value []byte + require.Zero(t, m.GetXattr(ctx, inode, "user.old", &value)) + require.Equal(t, []byte("original"), value) + require.Equal(t, ENOATTR, m.GetXattr(ctx, inode, "user.new", &value)) +} + +func testDirMarkerChangelog(t *testing.T, m Meta) { + ctx := Background() + var inode Ino + var attr Attr + require.Zero(t, m.Mkdir(ctx, RootInode, "marker-log", 0755, 0, 0, &inode, nil)) + updates := map[string][]byte{"user.binary": {0, 255, ',', '%'}, "user.deleted": nil} + require.Zero(t, m.SetDirMarker(ctx, RootInode, "marker-log", inode, true, updates, &attr)) + logCtx := NewContext(1, 0, []uint32{0}) + defer logCtx.Cancel() + timer := time.AfterFunc(5*time.Second, logCtx.Cancel) + defer timer.Stop() + done := errors.New("record found") + err := m.ScanChangelog(logCtx, 1, func(version int64, entry string) error { + parts := strings.Split(entry, "|") + if len(parts) < 2 || !strings.HasPrefix(parts[1], fmt.Sprintf("SETDIRMARKER(%d,", inode)) { + return nil + } + fields := strings.SplitN(strings.TrimSuffix(strings.TrimPrefix(parts[1], "SETDIRMARKER("), ")"), ",", 4) + require.Len(t, fields, 4) + require.Equal(t, strconv.FormatInt(attr.Ctime, 10), fields[1]) + require.Equal(t, strconv.FormatUint(uint64(attr.Ctimensec), 10), fields[2]) + raw, err := url.PathUnescape(fields[3]) + require.NoError(t, err) + var stored map[string][]byte + require.NoError(t, json.Unmarshal([]byte(raw), &stored)) + require.Equal(t, updates, stored) + return done + }) + require.ErrorIs(t, err, done) +} + +func testDirMarkerRedisFailure(t *testing.T, m *redisMeta) { + ctx := Background() + var inode Ino + var before, after Attr + require.Zero(t, m.Mkdir(ctx, RootInode, "marker-corrupt", 0755, 0, 0, &inode, &before)) + // Redis EXEC does not roll back runtime command errors. A wrong-type xattr + // key must be rejected before enqueuing either the marker or xattr writes. + require.NoError(t, m.rdb.Set(ctx, m.xattrKey(inode), "invalid hash", 0).Err()) + require.NotZero(t, m.SetDirMarker(ctx, RootInode, "marker-corrupt", inode, true, map[string][]byte{"user.new": []byte("value")}, &after)) + require.Zero(t, m.doGetAttr(ctx, inode, &after)) + require.Equal(t, before, after) + stored, err := m.rdb.Get(ctx, m.xattrKey(inode)).Result() + require.NoError(t, err) + require.Equal(t, "invalid hash", stored) + require.NoError(t, m.rdb.Del(ctx, m.xattrKey(inode)).Err()) +} + +func testDirMarkerWriters(t *testing.T, clients []Meta) { + m := clients[0] + ctx := Background() + name := strings.ReplaceAll(t.Name(), "/", "-") + var inode Ino + require.Zero(t, m.Mkdir(ctx, RootInode, name, 0755, 0, 0, &inode, nil)) + const writers = 16 + var wait sync.WaitGroup + start := make(chan struct{}) + results := make([]syscall.Errno, writers) + for i := range results { + wait.Add(1) + go func(i int) { + defer wait.Done() + <-start + var attr Attr + value := []byte(fmt.Sprint(i)) + results[i] = clients[i%len(clients)].SetDirMarker(ctx, RootInode, name, inode, true, map[string][]byte{"user.first": value, "user.second": value}, &attr) + }(i) + } + close(start) + wait.Wait() + winner := -1 + for i, st := range results { + if st == 0 { + require.Equal(t, -1, winner) + winner = i + } else { + require.Equal(t, syscall.EEXIST, st) + } + } + require.NotEqual(t, -1, winner) + + var first, second []byte + require.Zero(t, m.GetXattr(ctx, inode, "user.first", &first)) + require.Zero(t, m.GetXattr(ctx, inode, "user.second", &second)) + require.Equal(t, []byte(fmt.Sprint(winner)), first) + require.Equal(t, first, second) +} diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go index 0cd69c31e72c..66736aa31219 100644 --- a/pkg/meta/interface.go +++ b/pkg/meta/interface.go @@ -431,6 +431,11 @@ type Meta interface { GetAttr(ctx Context, inode Ino, attr *Attr) syscall.Errno // SetAttr updates the attributes for given node. SetAttr(ctx Context, inode Ino, set uint16, sggidclearmode uint8, attr *Attr) syscall.Errno + // SetDirMarker atomically marks an existing directory as an object (atime zero) + // and updates its xattrs. Nil values remove xattrs; absent names are unchanged. + // The parent/name entry must still refer to inode. If exclusive is true, an + // existing marker returns EEXIST without changes. A replaced entry returns EAGAIN. + SetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno // Check setting attr is allowed or not CheckSetAttr(ctx Context, inode Ino, set uint16, attr Attr) syscall.Errno // Truncate changes the length for given file. diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go index 9eb3744bfb6d..1b2f85bb8319 100644 --- a/pkg/meta/redis.go +++ b/pkg/meta/redis.go @@ -1404,6 +1404,54 @@ func (m *redisMeta) doSetAttr(ctx Context, inode Ino, set uint16, sugidclearmode }, m.inodeKey(inode))) } +func (m *redisMeta) doSetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno { + return errno(m.txn(ctx, func(tx *redis.Tx) error { + entry, err := tx.HGet(ctx, m.entryKey(parent), name).Bytes() + if err != nil { + return err + } + _, currentInode := m.parseEntry(entry) + if currentInode != inode { + return syscall.EAGAIN + } + values, err := tx.MGet(ctx, m.inodeKey(parent), m.inodeKey(inode)).Result() + if err != nil { + return err + } + if values[0] == nil || values[1] == nil { + return syscall.ENOENT + } + var parentAttr, current Attr + m.parseAttr([]byte(values[0].(string)), &parentAttr) + m.parseAttr([]byte(values[1].(string)), ¤t) + now := time.Now() + if st := m.prepareDirMarker(ctx, parent, inode, &parentAttr, ¤t, exclusive, now); st != 0 { + return st + } + // Validate the hash type before MULTI: Redis cannot roll back a command + // error inside EXEC. Reading also keeps corrupt xattrs from publishing a marker. + if _, err = tx.HLen(ctx, m.xattrKey(inode)).Result(); err != nil { + return err + } + _, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error { + for key, value := range xattrs { + if value == nil { + pipe.HDel(ctx, m.xattrKey(inode), key) + } else { + pipe.HSet(ctx, m.xattrKey(inode), key, value) + } + } + pipe.Set(ctx, m.inodeKey(inode), m.marshal(¤t), 0) + m.genLog(ctx, pipe, now, "%s", dirMarkerLog(inode, ¤t, xattrs)) + return nil + }) + if err == nil { + *attr = current + } + return err + }, m.inodeKey(inode), m.inodeKey(parent), m.entryKey(parent), m.xattrKey(inode))) +} + func (m *redisMeta) doReadlink(ctx Context, inode Ino, noatime bool) (atime int64, target []byte, err error) { if noatime { target, err = m.rdb.Get(ctx, m.symKey(inode)).Bytes() diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go index fb696879d144..ca7fa50803df 100644 --- a/pkg/meta/sql.go +++ b/pkg/meta/sql.go @@ -1251,6 +1251,12 @@ func (m *dbMeta) shouldRetry(err error) bool { strings.Contains(msg, "duplicate entry") || strings.Contains(msg, "error 1020 (hy000)") || strings.Contains(msg, "invalid connection") || strings.Contains(msg, "bad connection") || errors.Is(err, io.EOF) || strings.Contains(msg, "serialize access") // could not send data to client: No buffer space available case "postgres": + // PostgreSQL-compatible servers can use different error messages for + // serialization failures and deadlocks. Both require a new transaction. + var sqlErr interface{ SQLState() string } + if errors.As(err, &sqlErr) && (sqlErr.SQLState() == "40001" || sqlErr.SQLState() == "40P01") { + return true + } if e, ok := err.(interface{ SafeToRetry() bool }); ok { return e.SafeToRetry() } @@ -1585,6 +1591,51 @@ func (m *dbMeta) doSetAttr(ctx Context, inode Ino, set uint16, sugidclearmode ui }, inode)) } +func (m *dbMeta) doSetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno { + return errno(m.txn(func(s *xorm.Session) error { + parentNode, currentNode := node{Inode: parent}, node{Inode: inode} + if err := m.getNodesForUpdate(s, &parentNode, ¤tNode); err != nil { + return err + } + entry := edge{Parent: parent, Name: []byte(name)} + ok, err := s.ForUpdate().Get(&entry) + if err != nil { + return err + } + if !ok { + return syscall.ENOENT + } + if entry.Inode != inode { + return syscall.EAGAIN + } + var parentAttr, current Attr + m.parseAttr(&parentNode, &parentAttr) + m.parseAttr(¤tNode, ¤t) + now := time.Now() + if st := m.prepareDirMarker(ctx, parent, inode, &parentAttr, ¤t, exclusive, now); st != 0 { + return st + } + for key, value := range xattrs { + filter := &xattr{Inode: inode, Name: key} + if _, err = s.Delete(filter); err != nil { + return err + } + if value != nil { + if err = mustInsert(s, &xattr{Inode: inode, Name: key, Value: value}); err != nil { + return err + } + } + } + m.parseNode(¤t, ¤tNode) + if _, err = s.Cols("atime", "atimensec", "ctime", "ctimensec").Update(¤tNode, &node{Inode: inode}); err != nil { + return err + } + m.genLog(ctx, s, now.UnixNano(), "%s", dirMarkerLog(inode, ¤t, xattrs)) + *attr = current + return nil + }, parent, inode)) +} + func (m *dbMeta) appendSlice(s *xorm.Session, inode Ino, indx uint32, buf []byte) error { var r sql.Result var err error diff --git a/pkg/meta/sql_pg_test.go b/pkg/meta/sql_pg_test.go new file mode 100644 index 000000000000..b0ae4ad570d4 --- /dev/null +++ b/pkg/meta/sql_pg_test.go @@ -0,0 +1,51 @@ +//go:build !nopg +// +build !nopg + +/* + * JuiceFS, Copyright 2020 Juicedata, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package meta + +import ( + "fmt" + "testing" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/require" + "xorm.io/xorm" +) + +func TestPostgresRetrySQLState(t *testing.T) { + db, err := xorm.NewEngine("pgx", "postgres://localhost/unused?sslmode=disable") + require.NoError(t, err) + defer db.Close() + m := &dbMeta{db: db} + for _, tc := range []struct { + code string + retry bool + }{ + {"40001", true}, // serialization failure, including CockroachDB + {"40P01", true}, // deadlock + {"23503", false}, // foreign key violation + {"42501", false}, // insufficient privilege + } { + t.Run(tc.code, func(t *testing.T) { + err := &pgconn.PgError{Code: tc.code, Message: "server-specific transaction error"} + require.Equal(t, tc.retry, m.shouldRetry(err)) + require.Equal(t, tc.retry, m.shouldRetry(fmt.Errorf("wrapped: %w", err))) + }) + } +} diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go index bc7743a4e2f3..e5fb79e75177 100644 --- a/pkg/meta/tkv.go +++ b/pkg/meta/tkv.go @@ -1269,6 +1269,47 @@ func (m *kvMeta) doSetAttr(ctx Context, inode Ino, set uint16, sugidclearmode ui }, inode)) } +func (m *kvMeta) doSetDirMarker(ctx Context, parent Ino, name string, inode Ino, exclusive bool, xattrs map[string][]byte, attr *Attr) syscall.Errno { + for _, value := range xattrs { + if value != nil && len(value) == 0 && m.Name() == "tikv" { + return syscall.EINVAL + } + } + + return errno(m.txn(ctx, func(tx *kvTxn) error { + entry := tx.get(m.entryKey(parent, name)) + if entry == nil { + return syscall.ENOENT + } + _, currentInode := m.parseEntry(entry) + if currentInode != inode { + return syscall.EAGAIN + } + values := tx.gets(m.inodeKey(parent), m.inodeKey(inode)) + if values[0] == nil || values[1] == nil { + return syscall.ENOENT + } + var parentAttr, current Attr + m.parseAttr(values[0], &parentAttr) + m.parseAttr(values[1], ¤t) + now := time.Now() + if st := m.prepareDirMarker(ctx, parent, inode, &parentAttr, ¤t, exclusive, now); st != 0 { + return st + } + for key, value := range xattrs { + if value == nil { + tx.delete(m.xattrKey(inode, key)) + } else { + tx.set(m.xattrKey(inode, key), value) + } + } + tx.set(m.inodeKey(inode), m.marshal(¤t)) + m.genLog(tx, now, "%s", dirMarkerLog(inode, ¤t, xattrs)) + *attr = current + return nil + }, parent, inode)) +} + func (m *kvMeta) doTruncate(ctx Context, inode Ino, flags uint8, length uint64, delta *dirStat, attr *Attr, skipPermCheck bool) syscall.Errno { return errno(m.txn(ctx, func(tx *kvTxn) error { *delta = dirStat{}