Browse Source
mount: stream all filer mutations over single ordered gRPC stream (#8770)
mount: stream all filer mutations over single ordered gRPC stream (#8770)
* filer: add StreamMutateEntry bidi streaming RPC Add a bidirectional streaming RPC that carries all filer mutation types (create, update, delete, rename) over a single ordered stream. This eliminates per-request connection overhead for pipelined operations and guarantees mutation ordering within a stream. The server handler delegates each request to the existing unary handlers (CreateEntry, UpdateEntry, DeleteEntry) and uses a proxy stream adapter for rename operations to reuse StreamRenameEntry logic. The is_last field signals completion for multi-response operations (rename sends multiple events per request; create/update/delete always send exactly one response with is_last=true). * mount: add streaming mutation multiplexer (streamMutateMux) Implement a client-side multiplexer that routes all filer mutation RPCs (create, update, delete, rename) over a single bidirectional gRPC stream. Multiple goroutines submit requests through a send channel; a dedicated sendLoop serializes them on the stream; a recvLoop dispatches responses to waiting callers via per-request channels. Key features: - Lazy stream opening on first use - Automatic reconnection on stream failure - Permanent fallback to unary RPCs if filer returns Unimplemented - Monotonic request_id for response correlation - Multi-response support for rename operations (is_last signaling) The mux is initialized on WFS and closed during unmount cleanup. No call sites use it yet — wiring comes in subsequent commits. * mount: route CreateEntry and UpdateEntry through streaming mux Wire all CreateEntry call sites to use wfs.streamCreateEntry() which routes through the StreamMutateEntry stream when available, falling back to unary RPCs otherwise. Also wire Link's UpdateEntry calls through wfs.streamUpdateEntry(). Updated call sites: - flushMetadataToFiler (file flush after write) - Mkdir (directory creation) - Symlink (symbolic link creation) - createRegularFile non-deferred path (Mknod) - flushFileMetadata (periodic metadata flush) - Link (hard link: update source + create link + rollback) * mount: route UpdateEntry and DeleteEntry through streaming mux Wire remaining mutation call sites through the streaming mux: - saveEntry (Setattr/chmod/chown/utimes) → streamUpdateEntry - Unlink → streamDeleteEntry (replaces RemoveWithResponse) - Rmdir → streamDeleteEntry (replaces RemoveWithResponse) All filer mutations except Rename now go through StreamMutateEntry when the filer supports it, with automatic unary RPC fallback. * mount: route Rename through streaming mux Wire Rename to use streamMutate.Rename() when available, with fallback to the existing StreamRenameEntry unary stream. The streaming mux sends rename as a StreamRenameEntryRequest oneof variant. The server processes it through the existing rename logic and sends multiple StreamRenameEntryResponse events (one per moved entry), with is_last=true on the final response. All filer mutations now go through a single ordered stream. * mount: fix stream mux connection ownership WithGrpcClient(streamingMode=true) closes the gRPC connection when the callback returns, destroying the stream. Own the connection directly via pb.GrpcDial so it stays alive for the stream's lifetime. Close it explicitly in recvLoop on stream failure and in Close on shutdown. * mount: fix rename failure for deferred-create files Three fixes for rename operations over the streaming mux: 1. lookupEntry: fall back to local metadata store when filer returns "not found" for entries in uncached directories. Files created with deferFilerCreate=true exist only in the local leveldb store until flushed; lookupEntry skipped the local store when the parent directory had never been readdir'd, causing rename to fail with ENOENT. 2. Rename: wait for pending async flushes and force synchronous flush of dirty metadata before sending rename to the filer. Covers the writebackCache case where close() defers the flush to a background worker that may not complete before rename fires. 3. StreamMutateEntry: propagate rename errors from server to client. Add error/errno fields to StreamMutateEntryResponse so the mount can map filer errors to correct FUSE status codes instead of silently returning OK. Also fix the existing Rename error handler which could return fuse.OK on unrecognized errors. * mount: fix streaming mux error handling, sendLoop lifecycle, and fallback Address PR review comments: 1. Server: populate top-level Error/Errno on StreamMutateEntryResponse for create/update/delete errors, not just rename. Previously update errors were silently dropped and create/delete errors were only in nested response fields that the client didn't check. 2. Client: check nested error fields in CreateEntry (ErrorCode, Error) and DeleteEntry (Error) responses, matching CreateEntryWithResponse behavior. 3. Fix sendLoop lifecycle: give each stream generation a stopSend channel. recvLoop closes it on error to stop the paired sendLoop. Previously a reconnect left the old sendLoop draining sendCh, breaking ordering. 4. Transparent fallback: stream helpers and doRename fall back to unary RPCs on transport errors (ErrStreamTransport), including the first Unimplemented from ensureStream. Previously the first call failed instead of degrading. 5. Filer rotation in openStream: try all filer addresses on dial failure, matching WithFilerClient behavior. Stop early on Unimplemented. 6. Pass metadata-bearing context to StreamMutateEntry RPC call so sw-client-id header is actually sent. 7. Gate lookupEntry local-cache fallback on open dirty handle or pending async flush to avoid resurrecting deleted/renamed entries. 8. Remove dead code in flushFileMetadata (err=nil followed by if err!=nil). 9. Use string matching for rename error-to-errno mapping in the mount to stay portable across Linux/macOS (numeric errno values differ). * mount: make failAllPending idempotent with delete-before-close Change failAllPending to collect pending entries into a local slice (deleting from the sync.Map first) before closing channels. This prevents double-close panics if called concurrently. Also remove the unused err parameter. * mount: add stream generation tracking and teardownStream Introduce a generation counter on streamMutateMux that increments each time a new stream is created. Requests carry the generation they were enqueued for so sendLoop can reject stale requests after reconnect. Add teardownStream(gen) which is idempotent (only acts when gen matches current generation and stream is non-nil). Both sendLoop and recvLoop call it on error, replacing the inline cleanup in recvLoop. sendLoop now actively triggers teardown on send errors instead of silently exiting. ensureStream waits for the prior generation's recvDone before creating a new stream, ensuring all old pending waiters are failed before reconnect. recvLoop now takes the stream, generation, and recvDone channel as parameters to avoid accessing shared fields without the lock. * mount: harden Close to prevent races with teardownStream Nil out stream, cancel, and grpcConn under the lock so that any concurrent teardownStream call from recvLoop/sendLoop becomes a no-op. Call failAllPending before closing sendCh to unblock waiters promptly. Guard recvDone with a nil check for the case where Close is called before any stream was ever opened. * mount: make errCh receive ctx-aware in doUnary and Rename Replace the blocking <-sendReq.errCh with a select that also observes ctx.Done(). If sendLoop exits via stopSend without consuming a buffered request, the caller now returns ctx.Err() instead of blocking forever. The buffered errCh (capacity 1) ensures late acknowledgements from sendLoop don't block the sender. * mount: fix sendLoop/Close race and recvLoop/teardown pending channel race Three related fixes: 1. Stop closing sendCh in Close(). Closing the shared producer channel races with callers who passed ensureStream() but haven't sent yet, causing send-on-closed-channel panics. sendCh is now left open; ensureStream checks m.closed to reject new callers. 2. Drain buffered sendCh items on shutdown. sendLoop defers drainSendCh() on exit so buffered requests get an ErrStreamTransport on their errCh instead of blocking forever. Close() drains again for any stragglers enqueued between sendLoop's drain and the final shutdown. 3. Move failAllPending from teardownStream into recvLoop's defer. teardownStream (called from sendLoop on send error) was closing pending response channels while recvLoop could be between pending.Load and the channel send — a send-on-closed-channel panic. recvLoop is now the sole closer of pending channels, eliminating the race. Close() waits on recvDone (with cancel() to guarantee Recv unblocks) so pending cleanup always completes. * filer/mount: add debug logging for hardlink lifecycle Add V(0) logging at every point where a HardLinkId is created, stored, read, or deleted to trace orphaned hardlink references. Logging covers: - gRPC server: CreateEntry/UpdateEntry when request carries HardLinkId - FilerStoreWrapper: InsertEntry/UpdateEntry when entry has HardLinkId - handleUpdateToHardLinks: entry path, HardLinkId, counter, chunk count - setHardLink: KvPut with blob size - maybeReadHardLink: V(1) on read attempt and successful decode - DeleteHardLink: counter decrement/deletion events - Mount Link(): when NewHardLinkId is generated and link is created This helps diagnose how a git pack .rev file ended up with a HardLinkId during a clone (no hard links should be involved). * test: add git clone/pull integration test for FUSE mount Shell script that exercises git operations on a SeaweedFS mount: 1. Creates a bare repo on the mount 2. Clones locally, makes 3 commits, pushes to mount 3. Clones from mount bare repo into an on-mount working dir 4. Verifies clone integrity (files, content, commit hashes) 5. Pushes 2 more commits with renames and deletes 6. Checks out an older revision on the mount clone 7. Returns to branch and pulls with real changes 8. Verifies file content, renames, deletes after pull 9. Checks git log integrity and clean status 27 assertions covering file existence, content, commit hashes, file counts, renames, deletes, and git status. Run against any existing mount: bash test-git-on-mount.sh /path/to/mount * test: add git clone/pull FUSE integration test to CI suite Add TestGitOperations to the existing fuse_integration test framework. The test exercises git's full file operation surface on the mount: 1. Creates a bare repo on the mount (acts as remote) 2. Clones locally, makes 3 commits (files, bulk data, renames), pushes 3. Clones from mount bare repo into an on-mount working dir 4. Verifies clone integrity (content, commit hash, file count) 5. Pushes 2 more commits with new files, renames, and deletes 6. Checks out an older revision on the mount clone 7. Returns to branch and pulls with real fast-forward changes 8. Verifies post-pull state: content, renames, deletes, file counts 9. Checks git log integrity (5 commits) and clean status Runs automatically in the existing fuse-integration.yml CI workflow. * mount: fix permission check with uid/gid mapping The permission checks in createRegularFile() and Access() compared the caller's local uid/gid against the entry's filer-side uid/gid without applying the uid/gid mapper. With -map.uid 501:0, a directory created as uid 0 on the filer would not match the local caller uid 501, causing hasAccess() to fall through to "other" permission bits and reject write access (0755 → other has r-x, no w). Fix: map entry uid/gid from filer-space to local-space before the hasAccess() call so both sides are in the same namespace. This fixes rsync -a failing with "Permission denied" on mkstempat when using uid/gid mapping. * mount: fix Mkdir/Symlink returning filer-side uid/gid to kernel Mkdir and Symlink used `defer wfs.mapPbIdFromFilerToLocal(entry)` to restore local uid/gid, but `outputPbEntry` writes the kernel response before the function returns — so the kernel received filer-side uid/gid (e.g., 0:0). macFUSE then caches these and rejects subsequent child operations (mkdir, create) because the caller uid (501) doesn't match the directory owner (0), and "other" bits (0755 → r-x) lack write permission. Fix: replace the defer with an explicit call to mapPbIdFromFilerToLocal before outputPbEntry, so the kernel gets local uid/gid. Also add nil guards for UidGidMapper in Access and createRegularFile to prevent panics in tests that don't configure a mapper. This fixes rsync -a "Permission denied" on mkpathat for nested directories when using uid/gid mapping. * mount: fix Link outputting filer-side uid/gid to kernel, add nil guards Link had the same defer-before-outputPbEntry bug as Mkdir and Symlink: the kernel received filer-side uid/gid because the defer hadn't run yet when outputPbEntry wrote the response. Also add nil guards for UidGidMapper in Access and createRegularFile so tests without a mapper don't panic. Audit of all outputPbEntry/outputFilerEntry call sites: - Mkdir: fixed in prior commit (explicit map before output) - Symlink: fixed in prior commit (explicit map before output) - Link: fixed here (explicit map before output) - Create (existing file): entry from maybeLoadEntry (already mapped) - Create (deferred): entry has local uid/gid (never mapped to filer) - Create (non-deferred): createRegularFile defer runs before return - Mknod: createRegularFile defer runs before return - Lookup: entry from lookupEntry (already mapped) - GetAttr: entry from maybeReadEntry/maybeLoadEntry (already mapped) - readdir: entry from cache (mapIdFromFilerToLocal) or filer (mapped) - saveEntry: no kernel output - flushMetadataToFiler: no kernel output - flushFileMetadata: no kernel output * test: fix git test for same-filesystem FUSE clone When both the bare repo and working clone live on the same FUSE mount, git's local transport uses hardlinks and cross-repo stat calls that fail on FUSE. Fix: - Use --no-local on clone to disable local transport optimizations - Use reset --hard instead of checkout to stay on branch - Use fetch + reset --hard origin/<branch> instead of git pull to avoid local transport stat failures during fetch * adjust logging * test: use plain git clone/pull to exercise real FUSE behavior Remove --no-local and fetch+reset workarounds. The test should use the same git commands users run (clone, reset --hard, pull) so it reveals real FUSE issues rather than hiding them. * test: enable V(1) logging for filer/mount and collect logs on failure - Run filer and mount with -v=1 so hardlink lifecycle logs (V(0): create/delete/insert, V(1): read attempts) are captured - On test failure, automatically dump last 16KB of all process logs (master, volume, filer, mount) to test output - Copy process logs to /tmp/seaweedfs-fuse-logs/ for CI artifact upload - Update CI workflow to upload SeaweedFS process logs alongside test output * mount: clone entry for filer flush to prevent uid/gid race flushMetadataToFiler and flushFileMetadata used entry.GetEntry() which returns the file handle's live proto entry pointer, then mutated it in-place via mapPbIdFromLocalToFiler. During the gRPC call window, a concurrent Lookup (which takes entryLock.RLock but NOT fhLockTable) could observe filer-side uid/gid (e.g., 0:0) on the file handle entry and return it to the kernel. The kernel caches these attributes, so subsequent opens by the local user (uid 501) fail with EACCES. Fix: proto.Clone the entry before mapping uid/gid for the filer request. The file handle's live entry is never mutated, so concurrent Lookup always sees local uid/gid. This fixes the intermittent "Permission denied" on .git/FETCH_HEAD after the first git pull on a mount with uid/gid mapping. * mount: add debug logging for stale lock file investigation Add V(0) logging to trace the HEAD.lock recreation issue: - Create: log when O_EXCL fails (file already exists) with uid/gid/mode - completeAsyncFlush: log resolved path, saved path, dirtyMetadata, isDeleted at entry to trace whether async flush fires after rename - flushMetadataToFiler: log the dir/name/fullpath being flushed This will show whether the async flush is recreating the lock file after git renames HEAD.lock → HEAD. * mount: prevent async flush from recreating renamed .lock files When git renames HEAD.lock → HEAD, the async flush from the prior close() can run AFTER the rename and re-insert HEAD.lock into the meta cache via its CreateEntryRequest response event. The next git pull then sees HEAD.lock and fails with "File exists". Fix: add isRenamed flag on FileHandle, set by Rename before waiting for the pending async flush. The async flush checks this flag and skips the metadata flush for renamed files (same pattern as isDeleted for unlinked files). The data pages still flush normally. The Rename handler flushes deferred metadata synchronously (Case 1) before setting isRenamed, ensuring the entry exists on the filer for the rename to proceed. For already-released handles (Case 2), the entry was created by a prior flush. * mount: also mark renamed inodes via entry.Attributes.Inode fallback When GetInode fails (Forget already removed the inode mapping), the Rename handler couldn't find the pending async flush to set isRenamed. The async flush then recreated the .lock file on the filer. Fix: fall back to oldEntry.Attributes.Inode to find the pending async flush when the inode-to-path mapping is gone. Also extract MarkInodeRenamed into a method on FileHandleToInode for clarity. * mount: skip async metadata flush when saved path no longer maps to inode The isRenamed flag approach failed for refs/remotes/origin/HEAD.lock because neither GetInode nor oldEntry.Attributes.Inode could find the inode (Forget already evicted the mapping, and the entry's stored inode was 0). Add a direct check in completeAsyncFlush: before flushing metadata, verify that the saved path still maps to this inode in the inode-to-path table. If the path was renamed or removed (inode mismatch or not found), skip the metadata flush to avoid recreating a stale entry. This catches all rename cases regardless of whether the Rename handler could set the isRenamed flag. * mount: wait for pending async flush in Unlink before filer delete Unlink was deleting the filer entry first, then marking the draining async-flush handle as deleted. The async flush worker could race between these two operations and recreate the just-unlinked entry on the filer. This caused git's .lock files (e.g. refs/remotes/origin/HEAD.lock) to persist after git pull, breaking subsequent git operations. Move the isDeleted marking and add waitForPendingAsyncFlush() before the filer delete so any in-flight flush completes first. Even if the worker raced past the isDeleted check, the wait ensures it finishes before the filer delete cleans up any recreated entry. * mount: reduce async flush and metadata flush log verbosity Raise completeAsyncFlush entry log, saved-path-mismatch skip log, and flushMetadataToFiler entry log from V(0) to V(3)/V(4). These fire for every file close with writebackCache and are too noisy for normal use. * filer: reduce hardlink debug log verbosity from V(0) to V(4) HardLinkId logs in filerstore_wrapper, filerstore_hardlink, and filer_grpc_server fire on every hardlinked file operation (git pack files use hardlinks extensively) and produce excessive noise. * mount/filer: reduce noisy V(0) logs for link, rmdir, and empty folder check - weedfs_link.go: hardlink creation logs V(0) → V(4) - weedfs_dir_mkrm.go: non-empty folder rmdir error V(0) → V(1) - empty_folder_cleaner.go: "not empty" check log V(0) → V(4) * filer: handle missing hardlink KV as expected, not error A "kv: not found" on hardlink read is normal when the link blob was already cleaned up but a stale entry still references it. Log at V(1) for not-found; keep Error level for actual KV failures. * test: add waitForDir before git pull in FUSE git operations test After git reset --hard, the FUSE mount's metadata cache may need a moment to settle on slow CI. The git pull subprocess (unpack-objects) could fail to stat the working directory. Poll for up to 5s. * Update git_operations_test.go * wait * test: simplify FUSE test framework to use weed mini Replace the 4-process setup (master + volume + filer + mount) with 2 processes: "weed mini" (all-in-one) + "weed mount". This simplifies startup, reduces port allocation, and is faster on CI. * test: fix mini flag -admin → -admin.uipull/8773/head
Chris Lu
15 hours ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
30 changed files with 2331 additions and 469 deletions
-
1.github/workflows/fuse-integration.yml
-
260docker/compose/test-git-on-mount.sh
-
32other/java/client/src/main/proto/filer.proto
-
197test/fuse_integration/framework_test.go
-
348test/fuse_integration/git_operations_test.go
-
30test/fuse_integration/posix_file_lock_test.go
-
2weed/filer/empty_folder_cleanup/empty_folder_cleaner.go
-
27weed/filer/filerstore_hardlink.go
-
10weed/filer/filerstore_wrapper.go
-
1weed/mount/filehandle.go
-
11weed/mount/filehandle_map.go
-
27weed/mount/weedfs.go
-
8weed/mount/weedfs_access.go
-
21weed/mount/weedfs_async_flush.go
-
53weed/mount/weedfs_dir_mkrm.go
-
80weed/mount/weedfs_file_mkrm.go
-
93weed/mount/weedfs_file_sync.go
-
51weed/mount/weedfs_link.go
-
98weed/mount/weedfs_metadata_flush.go
-
124weed/mount/weedfs_rename.go
-
66weed/mount/weedfs_stream_helpers.go
-
506weed/mount/weedfs_stream_mutate.go
-
19weed/mount/weedfs_symlink.go
-
31weed/mount/wfs_save.go
-
32weed/pb/filer.proto
-
463weed/pb/filer_pb/filer.pb.go
-
38weed/pb/filer_pb/filer_grpc.pb.go
-
6weed/server/filer_grpc_server.go
-
161weed/server/filer_grpc_server_stream_mutate.go
-
4weed/storage/needle_map_leveldb.go
@ -0,0 +1,260 @@ |
|||
#!/usr/bin/env bash |
|||
# |
|||
# Integration test: git clone & pull on a SeaweedFS FUSE mount. |
|||
# |
|||
# Verifies that the mount correctly supports git's file operations by: |
|||
# 1. Creating a bare repo on the mount (acts as a remote) |
|||
# 2. Cloning it, making commits, and pushing back to the mount |
|||
# 3. Cloning from the mount into a working directory also on the mount |
|||
# 4. Pushing additional commits to the bare repo |
|||
# 5. Checking out an older revision in the on-mount clone |
|||
# 6. Running git pull to fast-forward with real changes |
|||
# 7. Verifying file content integrity at each step |
|||
# |
|||
# Usage: |
|||
# bash test-git-on-mount.sh /path/to/mount/point |
|||
# |
|||
# The mount must already be running. All test artifacts are created under |
|||
# <mount>/git-test-<pid> and cleaned up on exit (unless TEST_KEEP=1). |
|||
# |
|||
set -euo pipefail |
|||
|
|||
MOUNT_DIR="${1:?Usage: $0 <mount-dir>}" |
|||
TEST_DIR="$MOUNT_DIR/git-test-$$" |
|||
LOCAL_DIR=$(mktemp -d) |
|||
PASS=0 |
|||
FAIL=0 |
|||
|
|||
cleanup() { |
|||
if [[ "${TEST_KEEP:-}" == "1" ]]; then |
|||
echo "TEST_KEEP=1 — leaving artifacts:" |
|||
echo " mount: $TEST_DIR" |
|||
echo " local: $LOCAL_DIR" |
|||
else |
|||
rm -rf "$TEST_DIR" 2>/dev/null || true |
|||
rm -rf "$LOCAL_DIR" 2>/dev/null || true |
|||
fi |
|||
} |
|||
trap cleanup EXIT |
|||
|
|||
# --- helpers --------------------------------------------------------------- |
|||
|
|||
pass() { PASS=$((PASS + 1)); echo " PASS: $1"; } |
|||
fail() { FAIL=$((FAIL + 1)); echo " FAIL: $1"; } |
|||
|
|||
assert_file_contains() { |
|||
local file=$1 expected=$2 label=$3 |
|||
if [[ -f "$file" ]] && grep -qF "$expected" "$file" 2>/dev/null; then |
|||
pass "$label" |
|||
else |
|||
fail "$label (expected '$expected' in $file)" |
|||
fi |
|||
} |
|||
|
|||
assert_file_exists() { |
|||
local file=$1 label=$2 |
|||
if [[ -f "$file" ]]; then |
|||
pass "$label" |
|||
else |
|||
fail "$label ($file not found)" |
|||
fi |
|||
} |
|||
|
|||
assert_file_not_exists() { |
|||
local file=$1 label=$2 |
|||
if [[ ! -f "$file" ]]; then |
|||
pass "$label" |
|||
else |
|||
fail "$label ($file should not exist)" |
|||
fi |
|||
} |
|||
|
|||
assert_eq() { |
|||
local actual=$1 expected=$2 label=$3 |
|||
if [[ "$actual" == "$expected" ]]; then |
|||
pass "$label" |
|||
else |
|||
fail "$label (expected '$expected', got '$actual')" |
|||
fi |
|||
} |
|||
|
|||
# --- setup ----------------------------------------------------------------- |
|||
|
|||
echo "========================================" |
|||
echo " Git-on-mount integration test" |
|||
echo "========================================" |
|||
echo "Mount: $MOUNT_DIR" |
|||
echo "Test: $TEST_DIR" |
|||
echo "Local: $LOCAL_DIR" |
|||
echo "" |
|||
|
|||
if ! mountpoint -q "$MOUNT_DIR" 2>/dev/null && [[ ! -d "$MOUNT_DIR" ]]; then |
|||
echo "ERROR: $MOUNT_DIR is not a valid directory" |
|||
exit 1 |
|||
fi |
|||
|
|||
mkdir -p "$TEST_DIR" |
|||
|
|||
# --- Phase 1: Create bare repo on mount ----------------------------------- |
|||
|
|||
echo "--- Phase 1: Create bare repo on mount ---" |
|||
|
|||
BARE_REPO="$TEST_DIR/repo.git" |
|||
git init --bare "$BARE_REPO" >/dev/null 2>&1 |
|||
pass "bare repo created on mount" |
|||
|
|||
# --- Phase 2: Clone locally, make initial commits, push ------------------- |
|||
|
|||
echo "--- Phase 2: Clone locally, make initial commits, push ---" |
|||
|
|||
LOCAL_CLONE="$LOCAL_DIR/clone1" |
|||
git clone "$BARE_REPO" "$LOCAL_CLONE" >/dev/null 2>&1 |
|||
cd "$LOCAL_CLONE" |
|||
git config user.email "test@seaweedfs.test" |
|||
git config user.name "SeaweedFS Test" |
|||
|
|||
# Commit 1: initial files |
|||
echo "hello world" > README.md |
|||
mkdir -p src |
|||
echo 'package main; import "fmt"; func main() { fmt.Println("v1") }' > src/main.go |
|||
git add -A && git commit -m "initial commit" >/dev/null 2>&1 |
|||
COMMIT1=$(git rev-parse HEAD) |
|||
|
|||
# Commit 2: add more files |
|||
mkdir -p data |
|||
for i in $(seq 1 20); do |
|||
printf "file-%03d: %s\n" "$i" "$(head -c 64 /dev/urandom | base64)" > "data/file-$(printf '%03d' $i).txt" |
|||
done |
|||
git add -A && git commit -m "add data files" >/dev/null 2>&1 |
|||
COMMIT2=$(git rev-parse HEAD) |
|||
|
|||
# Commit 3: modify and add |
|||
echo 'package main; import "fmt"; func main() { fmt.Println("v2") }' > src/main.go |
|||
echo "# Updated readme" >> README.md |
|||
mkdir -p docs |
|||
echo "documentation content" > docs/guide.md |
|||
git add -A && git commit -m "update src and add docs" >/dev/null 2>&1 |
|||
COMMIT3=$(git rev-parse HEAD) |
|||
|
|||
git push origin master >/dev/null 2>&1 || git push origin main >/dev/null 2>&1 |
|||
BRANCH=$(git rev-parse --abbrev-ref HEAD) |
|||
pass "3 commits pushed to mount bare repo (branch=$BRANCH)" |
|||
|
|||
# --- Phase 3: Clone from mount bare repo to mount working dir ------------- |
|||
|
|||
echo "--- Phase 3: Clone from mount bare repo to on-mount working dir ---" |
|||
|
|||
MOUNT_CLONE="$TEST_DIR/working" |
|||
git clone "$BARE_REPO" "$MOUNT_CLONE" >/dev/null 2>&1 |
|||
|
|||
# Verify clone integrity |
|||
assert_file_exists "$MOUNT_CLONE/README.md" "README.md exists after clone" |
|||
assert_file_contains "$MOUNT_CLONE/README.md" "# Updated readme" "README.md has latest content" |
|||
assert_file_contains "$MOUNT_CLONE/src/main.go" 'v2' "src/main.go has v2" |
|||
assert_file_exists "$MOUNT_CLONE/docs/guide.md" "docs/guide.md exists" |
|||
assert_file_exists "$MOUNT_CLONE/data/file-001.txt" "data files exist" |
|||
assert_file_exists "$MOUNT_CLONE/data/file-020.txt" "data/file-020.txt exists" |
|||
|
|||
CLONE_HEAD=$(cd "$MOUNT_CLONE" && git rev-parse HEAD) |
|||
assert_eq "$CLONE_HEAD" "$COMMIT3" "on-mount clone HEAD matches commit 3" |
|||
|
|||
# Count files |
|||
FILE_COUNT=$(find "$MOUNT_CLONE/data" -name '*.txt' | wc -l | tr -d ' ') |
|||
assert_eq "$FILE_COUNT" "20" "data/ has 20 files" |
|||
|
|||
# --- Phase 4: Push more commits from local clone -------------------------- |
|||
|
|||
echo "--- Phase 4: Push more commits from local clone ---" |
|||
|
|||
cd "$LOCAL_CLONE" |
|||
|
|||
# Commit 4: larger changes |
|||
for i in $(seq 21 50); do |
|||
printf "file-%03d: %s\n" "$i" "$(head -c 128 /dev/urandom | base64)" > "data/file-$(printf '%03d' $i).txt" |
|||
done |
|||
echo 'package main; import "fmt"; func main() { fmt.Println("v3") }' > src/main.go |
|||
git add -A && git commit -m "expand data and update to v3" >/dev/null 2>&1 |
|||
COMMIT4=$(git rev-parse HEAD) |
|||
|
|||
# Commit 5: rename and delete |
|||
git mv docs/guide.md docs/manual.md |
|||
git rm data/file-001.txt >/dev/null 2>&1 |
|||
git commit -m "rename guide, remove file-001" >/dev/null 2>&1 |
|||
COMMIT5=$(git rev-parse HEAD) |
|||
|
|||
git push origin "$BRANCH" >/dev/null 2>&1 |
|||
pass "2 more commits pushed (5 total)" |
|||
|
|||
# --- Phase 5: Checkout older revision in on-mount clone ------------------- |
|||
|
|||
echo "--- Phase 5: Checkout older revision in on-mount clone ---" |
|||
|
|||
cd "$MOUNT_CLONE" |
|||
git checkout "$COMMIT2" >/dev/null 2>&1 |
|||
|
|||
# Verify we're at commit 2 state |
|||
DETACHED_HEAD=$(git rev-parse HEAD) |
|||
assert_eq "$DETACHED_HEAD" "$COMMIT2" "on-mount clone at commit 2 (detached)" |
|||
assert_file_not_exists "$MOUNT_CLONE/docs/guide.md" "docs/guide.md not in commit 2" |
|||
assert_file_contains "$MOUNT_CLONE/src/main.go" 'v1' "src/main.go has v1 at commit 2" |
|||
|
|||
# --- Phase 6: Return to branch and pull ----------------------------------- |
|||
|
|||
echo "--- Phase 6: Return to branch and pull with real changes ---" |
|||
|
|||
cd "$MOUNT_CLONE" |
|||
git checkout "$BRANCH" >/dev/null 2>&1 |
|||
# At this point the on-mount clone is at commit 3, remote is at commit 5 |
|||
OLD_HEAD=$(git rev-parse HEAD) |
|||
assert_eq "$OLD_HEAD" "$COMMIT3" "on-mount clone at commit 3 before pull" |
|||
|
|||
git pull >/dev/null 2>&1 |
|||
NEW_HEAD=$(git rev-parse HEAD) |
|||
assert_eq "$NEW_HEAD" "$COMMIT5" "HEAD matches commit 5 after pull" |
|||
|
|||
# Verify commit 5 state |
|||
assert_file_contains "$MOUNT_CLONE/src/main.go" 'v3' "src/main.go has v3 after pull" |
|||
assert_file_exists "$MOUNT_CLONE/docs/manual.md" "docs/manual.md exists (renamed)" |
|||
assert_file_not_exists "$MOUNT_CLONE/docs/guide.md" "docs/guide.md gone (renamed)" |
|||
assert_file_not_exists "$MOUNT_CLONE/data/file-001.txt" "data/file-001.txt removed" |
|||
assert_file_exists "$MOUNT_CLONE/data/file-050.txt" "data/file-050.txt exists" |
|||
|
|||
FINAL_COUNT=$(find "$MOUNT_CLONE/data" -name '*.txt' | wc -l | tr -d ' ') |
|||
assert_eq "$FINAL_COUNT" "49" "data/ has 49 files (50 added, 1 removed)" |
|||
|
|||
# --- Phase 7: Verify git log integrity ----------------------------------- |
|||
|
|||
echo "--- Phase 7: Verify git log integrity ---" |
|||
|
|||
cd "$MOUNT_CLONE" |
|||
GIT_LOG=$(git log --format=%s) |
|||
LOG_COUNT=$(echo "$GIT_LOG" | wc -l | tr -d ' ') |
|||
assert_eq "$LOG_COUNT" "5" "git log shows 5 commits" |
|||
|
|||
# Verify commit messages |
|||
echo "$GIT_LOG" | grep -qF "initial commit" && pass "commit 1 message in log" || fail "commit 1 message missing" |
|||
echo "$GIT_LOG" | grep -qF "expand data" && pass "commit 4 message in log" || fail "commit 4 message missing" |
|||
echo "$GIT_LOG" | grep -qF "rename guide" && pass "commit 5 message in log" || fail "commit 5 message missing" |
|||
|
|||
# --- Phase 8: Verify git status is clean ---------------------------------- |
|||
|
|||
echo "--- Phase 8: Verify git status is clean ---" |
|||
|
|||
cd "$MOUNT_CLONE" |
|||
STATUS=$(git status --porcelain) |
|||
if [[ -z "$STATUS" ]]; then |
|||
pass "git status is clean" |
|||
else |
|||
fail "git status has changes: $STATUS" |
|||
fi |
|||
|
|||
# --- Results --------------------------------------------------------------- |
|||
|
|||
echo "" |
|||
echo "========================================" |
|||
echo " Results: $PASS passed, $FAIL failed" |
|||
echo "========================================" |
|||
|
|||
if [[ "$FAIL" -gt 0 ]]; then |
|||
exit 1 |
|||
fi |
|||
@ -0,0 +1,348 @@ |
|||
package fuse_test |
|||
|
|||
import ( |
|||
"os" |
|||
"os/exec" |
|||
"path/filepath" |
|||
"regexp" |
|||
"strconv" |
|||
"strings" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/stretchr/testify/assert" |
|||
"github.com/stretchr/testify/require" |
|||
) |
|||
|
|||
// TestGitOperations exercises git clone, checkout, and pull on a FUSE mount.
|
|||
//
|
|||
// The test creates a bare repo on the mount (acting as a remote), clones it,
|
|||
// makes commits, pushes, then clones from the mount into an on-mount working
|
|||
// directory. It pushes additional commits, checks out an older revision in the
|
|||
// on-mount clone, and runs git pull to fast-forward with real changes —
|
|||
// verifying file content integrity at each step.
|
|||
func TestGitOperations(t *testing.T) { |
|||
framework := NewFuseTestFramework(t, DefaultTestConfig()) |
|||
defer framework.Cleanup() |
|||
|
|||
require.NoError(t, framework.Setup(DefaultTestConfig())) |
|||
|
|||
mountPoint := framework.GetMountPoint() |
|||
|
|||
// We need a local scratch dir (not on the mount) for the "developer" clone.
|
|||
localDir, err := os.MkdirTemp("", "git_ops_local_") |
|||
require.NoError(t, err) |
|||
defer os.RemoveAll(localDir) |
|||
|
|||
t.Run("CloneAndPull", func(t *testing.T) { |
|||
testGitCloneAndPull(t, mountPoint, localDir) |
|||
}) |
|||
} |
|||
|
|||
func testGitCloneAndPull(t *testing.T, mountPoint, localDir string) { |
|||
bareRepo := filepath.Join(mountPoint, "repo.git") |
|||
localClone := filepath.Join(localDir, "clone") |
|||
mountClone := filepath.Join(mountPoint, "working") |
|||
|
|||
// ---- Phase 1: Create bare repo on the mount ----
|
|||
t.Log("Phase 1: create bare repo on mount") |
|||
gitRun(t, "", "init", "--bare", bareRepo) |
|||
|
|||
// ---- Phase 2: Clone locally, make initial commits, push ----
|
|||
t.Log("Phase 2: clone locally, commit, push") |
|||
gitRun(t, "", "clone", bareRepo, localClone) |
|||
gitRun(t, localClone, "config", "user.email", "test@seaweedfs.test") |
|||
gitRun(t, localClone, "config", "user.name", "Test") |
|||
|
|||
// Commit 1
|
|||
writeFile(t, localClone, "README.md", "hello world\n") |
|||
mkdirAll(t, localClone, "src") |
|||
writeFile(t, localClone, "src/main.go", `package main; import "fmt"; func main() { fmt.Println("v1") }`) |
|||
gitRun(t, localClone, "add", "-A") |
|||
gitRun(t, localClone, "commit", "-m", "initial commit") |
|||
commit1 := gitOutput(t, localClone, "rev-parse", "HEAD") |
|||
|
|||
// Commit 2: bulk files
|
|||
mkdirAll(t, localClone, "data") |
|||
for i := 1; i <= 20; i++ { |
|||
name := filepath.Join("data", "file-"+leftPad(i, 3)+".txt") |
|||
writeFile(t, localClone, name, "content-"+strconv.Itoa(i)+"\n") |
|||
} |
|||
gitRun(t, localClone, "add", "-A") |
|||
gitRun(t, localClone, "commit", "-m", "add data files") |
|||
commit2 := gitOutput(t, localClone, "rev-parse", "HEAD") |
|||
|
|||
// Commit 3: modify + new dir
|
|||
writeFile(t, localClone, "src/main.go", `package main; import "fmt"; func main() { fmt.Println("v2") }`) |
|||
writeFile(t, localClone, "README.md", "hello world\n# Updated\n") |
|||
mkdirAll(t, localClone, "docs") |
|||
writeFile(t, localClone, "docs/guide.md", "documentation\n") |
|||
gitRun(t, localClone, "add", "-A") |
|||
gitRun(t, localClone, "commit", "-m", "update src and add docs") |
|||
commit3 := gitOutput(t, localClone, "rev-parse", "HEAD") |
|||
|
|||
branch := gitOutput(t, localClone, "rev-parse", "--abbrev-ref", "HEAD") |
|||
gitRun(t, localClone, "push", "origin", branch) |
|||
|
|||
// ---- Phase 3: Clone from mount bare repo into on-mount working dir ----
|
|||
t.Log("Phase 3: clone from mount bare repo to on-mount working dir") |
|||
gitRun(t, "", "clone", bareRepo, mountClone) |
|||
|
|||
assertFileContains(t, filepath.Join(mountClone, "README.md"), "# Updated") |
|||
assertFileContains(t, filepath.Join(mountClone, "src/main.go"), "v2") |
|||
assertFileExists(t, filepath.Join(mountClone, "docs/guide.md")) |
|||
assertFileExists(t, filepath.Join(mountClone, "data/file-020.txt")) |
|||
|
|||
head := gitOutput(t, mountClone, "rev-parse", "HEAD") |
|||
assert.Equal(t, commit3, head, "on-mount clone HEAD should be commit 3") |
|||
|
|||
dataFiles := countFiles(t, filepath.Join(mountClone, "data")) |
|||
assert.Equal(t, 20, dataFiles, "data/ should have 20 files") |
|||
|
|||
// ---- Phase 4: Push more commits from the local clone ----
|
|||
t.Log("Phase 4: push more commits") |
|||
|
|||
for i := 21; i <= 50; i++ { |
|||
name := filepath.Join("data", "file-"+leftPad(i, 3)+".txt") |
|||
writeFile(t, localClone, name, "content-"+strconv.Itoa(i)+"\n") |
|||
} |
|||
writeFile(t, localClone, "src/main.go", `package main; import "fmt"; func main() { fmt.Println("v3") }`) |
|||
gitRun(t, localClone, "add", "-A") |
|||
gitRun(t, localClone, "commit", "-m", "expand data and update to v3") |
|||
commit4 := gitOutput(t, localClone, "rev-parse", "HEAD") |
|||
_ = commit4 |
|||
|
|||
gitRun(t, localClone, "mv", "docs/guide.md", "docs/manual.md") |
|||
gitRun(t, localClone, "rm", "data/file-001.txt") |
|||
gitRun(t, localClone, "commit", "-m", "rename guide, remove file-001") |
|||
commit5 := gitOutput(t, localClone, "rev-parse", "HEAD") |
|||
|
|||
gitRun(t, localClone, "push", "origin", branch) |
|||
|
|||
// ---- Phase 5: Reset to older revision in on-mount clone ----
|
|||
t.Log("Phase 5: reset to older revision on mount clone") |
|||
ensureMountClone(t, bareRepo, mountClone) |
|||
gitRun(t, mountClone, "reset", "--hard", commit2) |
|||
|
|||
resetHead := gitOutput(t, mountClone, "rev-parse", "HEAD") |
|||
assert.Equal(t, commit2, resetHead, "should be at commit 2") |
|||
assertFileContains(t, filepath.Join(mountClone, "src/main.go"), "v1") |
|||
assertFileNotExists(t, filepath.Join(mountClone, "docs/guide.md")) |
|||
|
|||
// ---- Phase 6: Pull with real changes ----
|
|||
t.Log("Phase 6: pull with real fast-forward changes") |
|||
|
|||
ensureMountClone(t, bareRepo, mountClone) |
|||
|
|||
// After git reset --hard, give the FUSE mount a moment to settle its
|
|||
// metadata cache. On slow CI, the working directory can briefly appear
|
|||
// missing to a new subprocess (git pull → unpack-objects).
|
|||
waitForDir(t, mountClone) |
|||
|
|||
oldHead := gitOutput(t, mountClone, "rev-parse", "HEAD") |
|||
assert.Equal(t, commit2, oldHead, "should be at commit 2 before pull") |
|||
|
|||
gitRun(t, mountClone, "pull") |
|||
|
|||
newHead := gitOutput(t, mountClone, "rev-parse", "HEAD") |
|||
assert.Equal(t, commit5, newHead, "HEAD should be commit 5 after pull") |
|||
|
|||
assertFileContains(t, filepath.Join(mountClone, "src/main.go"), "v3") |
|||
assertFileExists(t, filepath.Join(mountClone, "docs/manual.md")) |
|||
assertFileNotExists(t, filepath.Join(mountClone, "docs/guide.md")) |
|||
assertFileNotExists(t, filepath.Join(mountClone, "data/file-001.txt")) |
|||
assertFileExists(t, filepath.Join(mountClone, "data/file-050.txt")) |
|||
|
|||
finalCount := countFiles(t, filepath.Join(mountClone, "data")) |
|||
assert.Equal(t, 49, finalCount, "data/ should have 49 files after pull") |
|||
|
|||
// ---- Phase 7: Verify git log and status ----
|
|||
t.Log("Phase 7: verify log and status") |
|||
logOutput := gitOutput(t, mountClone, "log", "--format=%s") |
|||
lines := strings.Split(strings.TrimSpace(logOutput), "\n") |
|||
assert.Equal(t, 5, len(lines), "should have 5 commits in log") |
|||
|
|||
assert.Contains(t, logOutput, "initial commit") |
|||
assert.Contains(t, logOutput, "expand data") |
|||
assert.Contains(t, logOutput, "rename guide") |
|||
|
|||
status := gitOutput(t, mountClone, "status", "--porcelain") |
|||
assert.Empty(t, status, "git status should be clean") |
|||
|
|||
_ = commit1 // used for documentation; not needed in assertions
|
|||
} |
|||
|
|||
// --- helpers ---
|
|||
|
|||
func gitRun(t *testing.T, dir string, args ...string) { |
|||
t.Helper() |
|||
gitRunWithRetry(t, dir, args...) |
|||
} |
|||
|
|||
func gitOutput(t *testing.T, dir string, args ...string) string { |
|||
t.Helper() |
|||
return gitRunWithRetry(t, dir, args...) |
|||
} |
|||
|
|||
// gitRunWithRetry runs a git command with retries to handle transient FUSE
|
|||
// I/O errors on slow CI runners (e.g. "Could not write new index file",
|
|||
// "failed to stat", "unpack-objects failed").
|
|||
func gitRunWithRetry(t *testing.T, dir string, args ...string) string { |
|||
t.Helper() |
|||
const ( |
|||
maxRetries = 6 |
|||
dirWait = 10 * time.Second |
|||
) |
|||
var out []byte |
|||
var err error |
|||
for i := 0; i < maxRetries; i++ { |
|||
if dir != "" && !waitForDirEventually(t, dir, dirWait) { |
|||
out = []byte("directory missing: " + dir) |
|||
err = &os.PathError{Op: "stat", Path: dir, Err: os.ErrNotExist} |
|||
} else { |
|||
cmd := exec.Command("git", args...) |
|||
if dir != "" { |
|||
cmd.Dir = dir |
|||
} |
|||
out, err = cmd.CombinedOutput() |
|||
} |
|||
if err == nil { |
|||
return strings.TrimSpace(string(out)) |
|||
} |
|||
if i < maxRetries-1 { |
|||
t.Logf("git %s attempt %d failed (retrying): %s", strings.Join(args, " "), i+1, string(out)) |
|||
if dir != "" { |
|||
refreshDirEntry(t, dir) |
|||
} |
|||
if repoPath := extractGitRepoPath(string(out)); repoPath != "" { |
|||
_ = exec.Command("git", "init", "--bare", repoPath).Run() |
|||
waitForBareRepoEventually(t, repoPath, 5*time.Second) |
|||
} |
|||
time.Sleep(500 * time.Millisecond) |
|||
} |
|||
} |
|||
require.NoError(t, err, "git %s failed after %d attempts: %s", strings.Join(args, " "), maxRetries, string(out)) |
|||
return "" |
|||
} |
|||
|
|||
func writeFile(t *testing.T, base, rel, content string) { |
|||
t.Helper() |
|||
p := filepath.Join(base, rel) |
|||
require.NoError(t, os.WriteFile(p, []byte(content), 0644)) |
|||
} |
|||
|
|||
func mkdirAll(t *testing.T, base, rel string) { |
|||
t.Helper() |
|||
require.NoError(t, os.MkdirAll(filepath.Join(base, rel), 0755)) |
|||
} |
|||
|
|||
func assertFileExists(t *testing.T, path string) { |
|||
t.Helper() |
|||
_, err := os.Stat(path) |
|||
require.NoError(t, err, "expected file to exist: %s", path) |
|||
} |
|||
|
|||
func assertFileNotExists(t *testing.T, path string) { |
|||
t.Helper() |
|||
_, err := os.Stat(path) |
|||
require.True(t, os.IsNotExist(err), "expected file not to exist: %s", path) |
|||
} |
|||
|
|||
func assertFileContains(t *testing.T, path, substr string) { |
|||
t.Helper() |
|||
data, err := os.ReadFile(path) |
|||
require.NoError(t, err, "failed to read %s", path) |
|||
assert.Contains(t, string(data), substr, "file %s should contain %q", path, substr) |
|||
} |
|||
|
|||
func countFiles(t *testing.T, dir string) int { |
|||
t.Helper() |
|||
entries, err := os.ReadDir(dir) |
|||
require.NoError(t, err, "failed to read dir %s", dir) |
|||
count := 0 |
|||
for _, e := range entries { |
|||
if !e.IsDir() { |
|||
count++ |
|||
} |
|||
} |
|||
return count |
|||
} |
|||
|
|||
func waitForDir(t *testing.T, dir string) { |
|||
t.Helper() |
|||
if !waitForDirEventually(t, dir, 10*time.Second) { |
|||
t.Fatalf("directory %s did not appear within 10s", dir) |
|||
} |
|||
} |
|||
|
|||
func waitForDirEventually(t *testing.T, dir string, timeout time.Duration) bool { |
|||
t.Helper() |
|||
deadline := time.Now().Add(timeout) |
|||
for time.Now().Before(deadline) { |
|||
if _, err := os.Stat(dir); err == nil { |
|||
return true |
|||
} |
|||
time.Sleep(100 * time.Millisecond) |
|||
} |
|||
return false |
|||
} |
|||
|
|||
func refreshDirEntry(t *testing.T, dir string) { |
|||
t.Helper() |
|||
parent := filepath.Dir(dir) |
|||
_, _ = os.ReadDir(parent) |
|||
} |
|||
|
|||
func waitForBareRepoEventually(t *testing.T, bareRepo string, timeout time.Duration) bool { |
|||
t.Helper() |
|||
deadline := time.Now().Add(timeout) |
|||
for time.Now().Before(deadline) { |
|||
if isBareRepo(bareRepo) { |
|||
return true |
|||
} |
|||
refreshDirEntry(t, bareRepo) |
|||
time.Sleep(150 * time.Millisecond) |
|||
} |
|||
return false |
|||
} |
|||
|
|||
func isBareRepo(bareRepo string) bool { |
|||
required := []string{ |
|||
filepath.Join(bareRepo, "HEAD"), |
|||
filepath.Join(bareRepo, "config"), |
|||
} |
|||
for _, p := range required { |
|||
if _, err := os.Stat(p); err != nil { |
|||
return false |
|||
} |
|||
} |
|||
return true |
|||
} |
|||
|
|||
func ensureMountClone(t *testing.T, bareRepo, mountClone string) { |
|||
t.Helper() |
|||
if _, err := os.Stat(mountClone); err == nil { |
|||
return |
|||
} else if !os.IsNotExist(err) { |
|||
require.NoError(t, err) |
|||
} |
|||
t.Logf("mount clone missing, re-cloning from %s", bareRepo) |
|||
gitRun(t, "", "clone", bareRepo, mountClone) |
|||
} |
|||
|
|||
var gitRepoPathRe = regexp.MustCompile(`'([^']+)' does not appear to be a git repository`) |
|||
|
|||
func extractGitRepoPath(output string) string { |
|||
if match := gitRepoPathRe.FindStringSubmatch(output); len(match) > 1 { |
|||
return match[1] |
|||
} |
|||
return "" |
|||
} |
|||
|
|||
func leftPad(n, width int) string { |
|||
s := strconv.Itoa(n) |
|||
for len(s) < width { |
|||
s = "0" + s |
|||
} |
|||
return s |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
package mount |
|||
|
|||
import ( |
|||
"context" |
|||
"errors" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/glog" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
) |
|||
|
|||
// streamCreateEntry routes a CreateEntryRequest through the streaming mux
|
|||
// if available, falling back to a unary gRPC call on transport errors.
|
|||
func (wfs *WFS) streamCreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { |
|||
if wfs.streamMutate != nil && wfs.streamMutate.IsAvailable() { |
|||
resp, err := wfs.streamMutate.CreateEntry(ctx, req) |
|||
if err == nil || !errors.Is(err, ErrStreamTransport) { |
|||
return resp, err // success or application error — don't retry
|
|||
} |
|||
glog.V(1).Infof("streamCreateEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Entry.Name, err) |
|||
} |
|||
var resp *filer_pb.CreateEntryResponse |
|||
err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { |
|||
var err error |
|||
resp, err = filer_pb.CreateEntryWithResponse(ctx, client, req) |
|||
return err |
|||
}) |
|||
return resp, err |
|||
} |
|||
|
|||
// streamUpdateEntry routes an UpdateEntryRequest through the streaming mux
|
|||
// if available, falling back to a unary gRPC call on transport errors.
|
|||
func (wfs *WFS) streamUpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { |
|||
if wfs.streamMutate != nil && wfs.streamMutate.IsAvailable() { |
|||
resp, err := wfs.streamMutate.UpdateEntry(ctx, req) |
|||
if err == nil || !errors.Is(err, ErrStreamTransport) { |
|||
return resp, err |
|||
} |
|||
glog.V(1).Infof("streamUpdateEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Entry.Name, err) |
|||
} |
|||
var resp *filer_pb.UpdateEntryResponse |
|||
err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { |
|||
var err error |
|||
resp, err = client.UpdateEntry(ctx, req) |
|||
return err |
|||
}) |
|||
return resp, err |
|||
} |
|||
|
|||
// streamDeleteEntry routes a DeleteEntryRequest through the streaming mux
|
|||
// if available, falling back to a unary gRPC call on transport errors.
|
|||
func (wfs *WFS) streamDeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { |
|||
if wfs.streamMutate != nil && wfs.streamMutate.IsAvailable() { |
|||
resp, err := wfs.streamMutate.DeleteEntry(ctx, req) |
|||
if err == nil || !errors.Is(err, ErrStreamTransport) { |
|||
return resp, err |
|||
} |
|||
glog.V(1).Infof("streamDeleteEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Name, err) |
|||
} |
|||
var resp *filer_pb.DeleteEntryResponse |
|||
err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { |
|||
var err error |
|||
resp, err = client.DeleteEntry(ctx, req) |
|||
return err |
|||
}) |
|||
return resp, err |
|||
} |
|||
@ -0,0 +1,506 @@ |
|||
package mount |
|||
|
|||
import ( |
|||
"context" |
|||
"errors" |
|||
"fmt" |
|||
"io" |
|||
"sync" |
|||
"sync/atomic" |
|||
"syscall" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/glog" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"google.golang.org/grpc" |
|||
"google.golang.org/grpc/codes" |
|||
grpcMetadata "google.golang.org/grpc/metadata" |
|||
"google.golang.org/grpc/status" |
|||
) |
|||
|
|||
// streamMutateError is returned when the server reports a structured errno.
|
|||
// It is also used by helpers to distinguish application errors (don't retry
|
|||
// on unary fallback) from transport errors (do retry).
|
|||
type streamMutateError struct { |
|||
msg string |
|||
errno syscall.Errno |
|||
} |
|||
|
|||
func (e *streamMutateError) Error() string { return e.msg } |
|||
func (e *streamMutateError) Errno() syscall.Errno { return e.errno } |
|||
|
|||
// ErrStreamTransport is a sentinel error type for transport-level stream
|
|||
// failures (disconnects, send errors). Callers use errors.Is to decide
|
|||
// whether to fall back to unary RPCs.
|
|||
var ErrStreamTransport = errors.New("stream transport error") |
|||
|
|||
// streamMutateMux multiplexes filer mutation RPCs (create, update, delete,
|
|||
// rename) over a single bidirectional gRPC stream. Multiple goroutines can
|
|||
// call the mutation methods concurrently; requests are serialized through
|
|||
// sendCh and responses are dispatched back via per-request channels.
|
|||
type streamMutateMux struct { |
|||
wfs *WFS |
|||
|
|||
mu sync.Mutex // protects stream, cancel, grpcConn, closed, stopSend, generation
|
|||
stream filer_pb.SeaweedFiler_StreamMutateEntryClient |
|||
cancel context.CancelFunc |
|||
grpcConn *grpc.ClientConn // dedicated connection, closed on stream teardown
|
|||
closed bool |
|||
disabled bool // permanently disabled if filer doesn't support the RPC
|
|||
stopSend chan struct{} // closed to signal the current sendLoop to exit
|
|||
generation uint64 // incremented each time a new stream is created
|
|||
|
|||
nextID atomic.Uint64 |
|||
|
|||
// pending maps request_id → response channel. The recvLoop dispatches
|
|||
// each response to the correct waiter. For rename (multi-response),
|
|||
// the channel receives multiple messages until is_last=true.
|
|||
pending sync.Map // map[uint64]chan *filer_pb.StreamMutateEntryResponse
|
|||
|
|||
sendCh chan *streamMutateReq |
|||
recvDone chan struct{} // closed when recvLoop exits
|
|||
} |
|||
|
|||
type streamMutateReq struct { |
|||
req *filer_pb.StreamMutateEntryRequest |
|||
errCh chan error // send error feedback
|
|||
gen uint64 // stream generation this request targets
|
|||
} |
|||
|
|||
func newStreamMutateMux(wfs *WFS) *streamMutateMux { |
|||
return &streamMutateMux{ |
|||
wfs: wfs, |
|||
sendCh: make(chan *streamMutateReq, 512), |
|||
} |
|||
} |
|||
|
|||
// CreateEntry sends a CreateEntryRequest over the stream and waits for the response.
|
|||
func (m *streamMutateMux) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { |
|||
resp, err := m.doUnary(ctx, &filer_pb.StreamMutateEntryRequest{ |
|||
Request: &filer_pb.StreamMutateEntryRequest_CreateRequest{CreateRequest: req}, |
|||
}) |
|||
if err != nil { |
|||
return nil, err |
|||
} |
|||
r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_CreateResponse) |
|||
if !ok { |
|||
return nil, fmt.Errorf("unexpected response type %T", resp.Response) |
|||
} |
|||
// Check nested error fields (same logic as CreateEntryWithResponse).
|
|||
cr := r.CreateResponse |
|||
if cr.ErrorCode != filer_pb.FilerError_OK { |
|||
if sentinel := filer_pb.FilerErrorToSentinel(cr.ErrorCode); sentinel != nil { |
|||
return nil, fmt.Errorf("CreateEntry %s/%s: %w", req.Directory, req.Entry.Name, sentinel) |
|||
} |
|||
return nil, &streamMutateError{msg: cr.Error, errno: syscall.EIO} |
|||
} |
|||
if cr.Error != "" { |
|||
return nil, &streamMutateError{msg: cr.Error, errno: syscall.EIO} |
|||
} |
|||
return cr, nil |
|||
} |
|||
|
|||
// UpdateEntry sends an UpdateEntryRequest over the stream and waits for the response.
|
|||
func (m *streamMutateMux) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { |
|||
resp, err := m.doUnary(ctx, &filer_pb.StreamMutateEntryRequest{ |
|||
Request: &filer_pb.StreamMutateEntryRequest_UpdateRequest{UpdateRequest: req}, |
|||
}) |
|||
if err != nil { |
|||
return nil, err |
|||
} |
|||
if r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_UpdateResponse); ok { |
|||
return r.UpdateResponse, nil |
|||
} |
|||
return nil, fmt.Errorf("unexpected response type %T", resp.Response) |
|||
} |
|||
|
|||
// DeleteEntry sends a DeleteEntryRequest over the stream and waits for the response.
|
|||
func (m *streamMutateMux) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { |
|||
resp, err := m.doUnary(ctx, &filer_pb.StreamMutateEntryRequest{ |
|||
Request: &filer_pb.StreamMutateEntryRequest_DeleteRequest{DeleteRequest: req}, |
|||
}) |
|||
if err != nil { |
|||
return nil, err |
|||
} |
|||
r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_DeleteResponse) |
|||
if !ok { |
|||
return nil, fmt.Errorf("unexpected response type %T", resp.Response) |
|||
} |
|||
// Check nested error field.
|
|||
if r.DeleteResponse.Error != "" { |
|||
return nil, &streamMutateError{msg: r.DeleteResponse.Error, errno: syscall.EIO} |
|||
} |
|||
return r.DeleteResponse, nil |
|||
} |
|||
|
|||
// Rename sends a StreamRenameEntryRequest over the stream and collects all
|
|||
// response events until is_last=true. The callback is invoked for each
|
|||
// intermediate rename event (same as the current StreamRenameEntry recv loop).
|
|||
func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRenameEntryRequest, onEvent func(*filer_pb.StreamRenameEntryResponse) error) error { |
|||
gen, err := m.ensureStream() |
|||
if err != nil { |
|||
return fmt.Errorf("%w: %v", ErrStreamTransport, err) |
|||
} |
|||
|
|||
id := m.nextID.Add(1) |
|||
ch := make(chan *filer_pb.StreamMutateEntryResponse, 64) |
|||
m.pending.Store(id, ch) |
|||
defer m.pending.Delete(id) |
|||
|
|||
sendReq := &streamMutateReq{ |
|||
req: &filer_pb.StreamMutateEntryRequest{ |
|||
RequestId: id, |
|||
Request: &filer_pb.StreamMutateEntryRequest_RenameRequest{RenameRequest: req}, |
|||
}, |
|||
errCh: make(chan error, 1), |
|||
gen: gen, |
|||
} |
|||
select { |
|||
case m.sendCh <- sendReq: |
|||
case <-ctx.Done(): |
|||
return ctx.Err() |
|||
} |
|||
select { |
|||
case err := <-sendReq.errCh: |
|||
if err != nil { |
|||
return fmt.Errorf("rename send: %w: %v", ErrStreamTransport, err) |
|||
} |
|||
case <-ctx.Done(): |
|||
return ctx.Err() |
|||
} |
|||
|
|||
// Collect rename events until is_last=true.
|
|||
for { |
|||
select { |
|||
case resp, ok := <-ch: |
|||
if !ok { |
|||
return fmt.Errorf("rename recv: %w: stream closed", ErrStreamTransport) |
|||
} |
|||
if r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_RenameResponse); ok { |
|||
if r.RenameResponse != nil && r.RenameResponse.EventNotification != nil { |
|||
if err := onEvent(r.RenameResponse); err != nil { |
|||
return err |
|||
} |
|||
} |
|||
} |
|||
if resp.IsLast { |
|||
if resp.Error != "" { |
|||
return &streamMutateError{ |
|||
msg: resp.Error, |
|||
errno: syscall.Errno(resp.Errno), |
|||
} |
|||
} |
|||
return nil |
|||
} |
|||
case <-ctx.Done(): |
|||
return ctx.Err() |
|||
} |
|||
} |
|||
} |
|||
|
|||
// doUnary sends a single-response request and waits for the reply.
|
|||
func (m *streamMutateMux) doUnary(ctx context.Context, req *filer_pb.StreamMutateEntryRequest) (*filer_pb.StreamMutateEntryResponse, error) { |
|||
gen, err := m.ensureStream() |
|||
if err != nil { |
|||
return nil, fmt.Errorf("%w: %v", ErrStreamTransport, err) |
|||
} |
|||
|
|||
id := m.nextID.Add(1) |
|||
req.RequestId = id |
|||
ch := make(chan *filer_pb.StreamMutateEntryResponse, 1) |
|||
m.pending.Store(id, ch) |
|||
defer m.pending.Delete(id) |
|||
|
|||
sendReq := &streamMutateReq{ |
|||
req: req, |
|||
errCh: make(chan error, 1), |
|||
gen: gen, |
|||
} |
|||
select { |
|||
case m.sendCh <- sendReq: |
|||
case <-ctx.Done(): |
|||
return nil, ctx.Err() |
|||
} |
|||
select { |
|||
case err := <-sendReq.errCh: |
|||
if err != nil { |
|||
return nil, fmt.Errorf("%w: %v", ErrStreamTransport, err) |
|||
} |
|||
case <-ctx.Done(): |
|||
return nil, ctx.Err() |
|||
} |
|||
|
|||
select { |
|||
case resp, ok := <-ch: |
|||
if !ok { |
|||
return nil, fmt.Errorf("%w: stream closed", ErrStreamTransport) |
|||
} |
|||
if resp.Error != "" { |
|||
return nil, &streamMutateError{ |
|||
msg: resp.Error, |
|||
errno: syscall.Errno(resp.Errno), |
|||
} |
|||
} |
|||
return resp, nil |
|||
case <-ctx.Done(): |
|||
return nil, ctx.Err() |
|||
} |
|||
} |
|||
|
|||
// ensureStream opens the bidi stream if not already open. It returns the
|
|||
// stream generation so callers can tag outgoing requests.
|
|||
func (m *streamMutateMux) ensureStream() (uint64, error) { |
|||
m.mu.Lock() |
|||
defer m.mu.Unlock() |
|||
|
|||
if m.closed { |
|||
return 0, fmt.Errorf("stream mux is closed") |
|||
} |
|||
if m.disabled { |
|||
return 0, fmt.Errorf("StreamMutateEntry not supported by filer") |
|||
} |
|||
if m.stream != nil { |
|||
return m.generation, nil |
|||
} |
|||
|
|||
// Wait for prior generation's recvLoop to fully tear down before opening
|
|||
// a new stream. This guarantees all pending waiters from the old stream
|
|||
// have been failed before we create a new generation.
|
|||
if m.recvDone != nil { |
|||
done := m.recvDone |
|||
m.mu.Unlock() |
|||
<-done |
|||
m.mu.Lock() |
|||
// Re-check after reacquiring the lock.
|
|||
if m.closed { |
|||
return 0, fmt.Errorf("stream mux is closed") |
|||
} |
|||
if m.disabled { |
|||
return 0, fmt.Errorf("StreamMutateEntry not supported by filer") |
|||
} |
|||
if m.stream != nil { |
|||
return m.generation, nil |
|||
} |
|||
} |
|||
|
|||
var stream filer_pb.SeaweedFiler_StreamMutateEntryClient |
|||
err := m.openStream(&stream) |
|||
if err != nil { |
|||
if s, ok := status.FromError(err); ok && s.Code() == codes.Unimplemented { |
|||
m.disabled = true |
|||
glog.V(0).Infof("filer does not support StreamMutateEntry, falling back to unary RPCs") |
|||
} |
|||
return 0, err |
|||
} |
|||
|
|||
m.generation++ |
|||
m.stream = stream |
|||
m.stopSend = make(chan struct{}) |
|||
recvDone := make(chan struct{}) |
|||
m.recvDone = recvDone |
|||
gen := m.generation |
|||
go m.sendLoop(stream, m.stopSend, gen) |
|||
go m.recvLoop(stream, gen, recvDone) |
|||
return gen, nil |
|||
} |
|||
|
|||
func (m *streamMutateMux) openStream(out *filer_pb.SeaweedFiler_StreamMutateEntryClient) error { |
|||
i := atomic.LoadInt32(&m.wfs.option.filerIndex) |
|||
n := int32(len(m.wfs.option.FilerAddresses)) |
|||
var lastErr error |
|||
|
|||
for x := int32(0); x < n; x++ { |
|||
idx := (i + x) % n |
|||
filerGrpcAddress := m.wfs.option.FilerAddresses[idx].ToGrpcAddress() |
|||
|
|||
ctx := context.Background() |
|||
if m.wfs.signature != 0 { |
|||
ctx = grpcMetadata.AppendToOutgoingContext(ctx, "sw-client-id", fmt.Sprintf("%d", m.wfs.signature)) |
|||
} |
|||
grpcConn, err := pb.GrpcDial(ctx, filerGrpcAddress, false, m.wfs.option.GrpcDialOption) |
|||
if err != nil { |
|||
lastErr = fmt.Errorf("stream dial %s: %v", filerGrpcAddress, err) |
|||
continue |
|||
} |
|||
|
|||
client := filer_pb.NewSeaweedFilerClient(grpcConn) |
|||
streamCtx, cancel := context.WithCancel(ctx) |
|||
stream, err := client.StreamMutateEntry(streamCtx) |
|||
if err != nil { |
|||
cancel() |
|||
grpcConn.Close() |
|||
lastErr = err |
|||
// Unimplemented means all filers lack it — stop rotating.
|
|||
if s, ok := status.FromError(err); ok && s.Code() == codes.Unimplemented { |
|||
return err |
|||
} |
|||
continue |
|||
} |
|||
|
|||
atomic.StoreInt32(&m.wfs.option.filerIndex, idx) |
|||
m.cancel = cancel |
|||
m.grpcConn = grpcConn |
|||
*out = stream |
|||
return nil |
|||
} |
|||
return lastErr |
|||
} |
|||
|
|||
func (m *streamMutateMux) sendLoop(stream filer_pb.SeaweedFiler_StreamMutateEntryClient, stop <-chan struct{}, gen uint64) { |
|||
defer m.drainSendCh() |
|||
for { |
|||
select { |
|||
case req, ok := <-m.sendCh: |
|||
if !ok { |
|||
return // defensive: sendCh should not be closed
|
|||
} |
|||
if req.gen != gen { |
|||
req.errCh <- fmt.Errorf("%w: stream generation mismatch", ErrStreamTransport) |
|||
continue |
|||
} |
|||
err := stream.Send(req.req) |
|||
req.errCh <- err |
|||
if err != nil { |
|||
m.teardownStream(gen) |
|||
return |
|||
} |
|||
case <-stop: |
|||
return |
|||
} |
|||
} |
|||
} |
|||
|
|||
func (m *streamMutateMux) recvLoop(stream filer_pb.SeaweedFiler_StreamMutateEntryClient, gen uint64, recvDone chan struct{}) { |
|||
defer func() { |
|||
m.failAllPending() |
|||
close(recvDone) |
|||
}() |
|||
for { |
|||
resp, err := stream.Recv() |
|||
if err != nil { |
|||
if err != io.EOF { |
|||
glog.V(1).Infof("stream mutate recv error (gen=%d): %v", gen, err) |
|||
} |
|||
m.teardownStream(gen) |
|||
return |
|||
} |
|||
|
|||
if ch, ok := m.pending.Load(resp.RequestId); ok { |
|||
ch.(chan *filer_pb.StreamMutateEntryResponse) <- resp |
|||
// For single-response ops, the caller deletes from pending after recv.
|
|||
// For rename, the caller collects until is_last.
|
|||
} |
|||
} |
|||
} |
|||
|
|||
// teardownStream cleans up the stream for the given generation. It is safe to
|
|||
// call from both sendLoop and recvLoop; only the first call for a given
|
|||
// generation takes effect (idempotent via generation + nil-stream check).
|
|||
func (m *streamMutateMux) teardownStream(gen uint64) { |
|||
m.mu.Lock() |
|||
if m.generation != gen || m.stream == nil { |
|||
m.mu.Unlock() |
|||
return |
|||
} |
|||
m.stream = nil |
|||
if m.stopSend != nil { |
|||
close(m.stopSend) |
|||
m.stopSend = nil |
|||
} |
|||
if m.cancel != nil { |
|||
m.cancel() |
|||
m.cancel = nil |
|||
} |
|||
conn := m.grpcConn |
|||
m.grpcConn = nil |
|||
m.mu.Unlock() |
|||
|
|||
// Do NOT call failAllPending here — recvLoop is the sole owner of
|
|||
// pending channel teardown. This avoids a race where teardownStream
|
|||
// closes a channel that recvLoop is about to send on.
|
|||
if conn != nil { |
|||
conn.Close() |
|||
} |
|||
} |
|||
|
|||
// failAllPending closes all pending response channels, causing waiters to
|
|||
// receive ok=false. It is idempotent: entries are deleted before channels are
|
|||
// closed, so concurrent calls cannot double-close.
|
|||
func (m *streamMutateMux) failAllPending() { |
|||
var channels []chan *filer_pb.StreamMutateEntryResponse |
|||
m.pending.Range(func(key, value any) bool { |
|||
m.pending.Delete(key) |
|||
channels = append(channels, value.(chan *filer_pb.StreamMutateEntryResponse)) |
|||
return true |
|||
}) |
|||
for _, ch := range channels { |
|||
close(ch) |
|||
} |
|||
} |
|||
|
|||
// drainSendCh drains buffered requests from sendCh, sending an error to each
|
|||
// request's errCh so callers don't block. Called by sendLoop's defer on exit
|
|||
// and by Close for any stragglers.
|
|||
func (m *streamMutateMux) drainSendCh() { |
|||
for { |
|||
select { |
|||
case req, ok := <-m.sendCh: |
|||
if !ok { |
|||
return // defensive: sendCh should not be closed
|
|||
} |
|||
req.errCh <- fmt.Errorf("%w: stream shutting down", ErrStreamTransport) |
|||
default: |
|||
return |
|||
} |
|||
} |
|||
} |
|||
|
|||
// IsAvailable returns true if the stream mux is usable (not permanently disabled).
|
|||
func (m *streamMutateMux) IsAvailable() bool { |
|||
m.mu.Lock() |
|||
defer m.mu.Unlock() |
|||
return !m.disabled |
|||
} |
|||
|
|||
// Close shuts down the stream. Called during unmount after all flushes complete.
|
|||
func (m *streamMutateMux) Close() { |
|||
m.mu.Lock() |
|||
if m.closed { |
|||
m.mu.Unlock() |
|||
return |
|||
} |
|||
m.closed = true |
|||
stream := m.stream |
|||
m.stream = nil // prevent teardownStream from acting after Close
|
|||
cancel := m.cancel |
|||
m.cancel = nil |
|||
grpcConn := m.grpcConn |
|||
m.grpcConn = nil |
|||
recvDone := m.recvDone |
|||
if m.stopSend != nil { |
|||
close(m.stopSend) |
|||
m.stopSend = nil |
|||
} |
|||
m.mu.Unlock() |
|||
|
|||
// CloseSend triggers EOF on recvLoop; cancel ensures Recv unblocks
|
|||
// even if the transport is broken.
|
|||
if stream != nil { |
|||
stream.CloseSend() |
|||
} |
|||
if cancel != nil { |
|||
cancel() |
|||
} |
|||
// Wait for recvLoop to finish — it calls failAllPending on exit.
|
|||
if recvDone != nil { |
|||
<-recvDone |
|||
} |
|||
if grpcConn != nil { |
|||
grpcConn.Close() |
|||
} |
|||
// Drain any remaining requests buffered in sendCh. sendLoop's defer
|
|||
// drain handles most items, but stragglers enqueued during shutdown
|
|||
// (between ensureStream and the sendCh send) are caught here.
|
|||
// sendCh is intentionally left open to prevent send-on-closed panics.
|
|||
m.drainSendCh() |
|||
} |
|||
@ -0,0 +1,161 @@ |
|||
package weed_server |
|||
|
|||
import ( |
|||
"context" |
|||
"io" |
|||
"strings" |
|||
"syscall" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/glog" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"google.golang.org/grpc" |
|||
"google.golang.org/grpc/metadata" |
|||
) |
|||
|
|||
func (fs *FilerServer) StreamMutateEntry(stream grpc.BidiStreamingServer[filer_pb.StreamMutateEntryRequest, filer_pb.StreamMutateEntryResponse]) error { |
|||
for { |
|||
req, err := stream.Recv() |
|||
if err == io.EOF { |
|||
return nil |
|||
} |
|||
if err != nil { |
|||
return err |
|||
} |
|||
|
|||
switch r := req.Request.(type) { |
|||
|
|||
case *filer_pb.StreamMutateEntryRequest_CreateRequest: |
|||
resp, createErr := fs.CreateEntry(stream.Context(), r.CreateRequest) |
|||
if createErr != nil { |
|||
resp = &filer_pb.CreateEntryResponse{Error: createErr.Error()} |
|||
} |
|||
streamResp := &filer_pb.StreamMutateEntryResponse{ |
|||
RequestId: req.RequestId, |
|||
IsLast: true, |
|||
Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{CreateResponse: resp}, |
|||
} |
|||
if resp.Error != "" { |
|||
streamResp.Error = resp.Error |
|||
streamResp.Errno = int32(syscall.EIO) |
|||
} |
|||
if sendErr := stream.Send(streamResp); sendErr != nil { |
|||
return sendErr |
|||
} |
|||
|
|||
case *filer_pb.StreamMutateEntryRequest_UpdateRequest: |
|||
resp, updateErr := fs.UpdateEntry(stream.Context(), r.UpdateRequest) |
|||
if updateErr != nil { |
|||
resp = &filer_pb.UpdateEntryResponse{} |
|||
} |
|||
streamResp := &filer_pb.StreamMutateEntryResponse{ |
|||
RequestId: req.RequestId, |
|||
IsLast: true, |
|||
Response: &filer_pb.StreamMutateEntryResponse_UpdateResponse{UpdateResponse: resp}, |
|||
} |
|||
if updateErr != nil { |
|||
streamResp.Error = updateErr.Error() |
|||
streamResp.Errno = int32(syscall.EIO) |
|||
} |
|||
if sendErr := stream.Send(streamResp); sendErr != nil { |
|||
return sendErr |
|||
} |
|||
|
|||
case *filer_pb.StreamMutateEntryRequest_DeleteRequest: |
|||
resp, deleteErr := fs.DeleteEntry(stream.Context(), r.DeleteRequest) |
|||
if deleteErr != nil { |
|||
resp = &filer_pb.DeleteEntryResponse{Error: deleteErr.Error()} |
|||
} |
|||
streamResp := &filer_pb.StreamMutateEntryResponse{ |
|||
RequestId: req.RequestId, |
|||
IsLast: true, |
|||
Response: &filer_pb.StreamMutateEntryResponse_DeleteResponse{DeleteResponse: resp}, |
|||
} |
|||
if resp.Error != "" { |
|||
streamResp.Error = resp.Error |
|||
streamResp.Errno = int32(syscall.EIO) |
|||
} |
|||
if sendErr := stream.Send(streamResp); sendErr != nil { |
|||
return sendErr |
|||
} |
|||
|
|||
case *filer_pb.StreamMutateEntryRequest_RenameRequest: |
|||
if err := fs.handleStreamMutateRename(stream, req.RequestId, r.RenameRequest); err != nil { |
|||
return err |
|||
} |
|||
|
|||
default: |
|||
glog.Warningf("StreamMutateEntry: unknown request type %T", req.Request) |
|||
} |
|||
} |
|||
} |
|||
|
|||
// handleStreamMutateRename delegates to the existing StreamRenameEntry logic
|
|||
// using a proxy stream that converts StreamRenameEntryResponse events into
|
|||
// StreamMutateEntryResponse messages on the parent bidi stream.
|
|||
func (fs *FilerServer) handleStreamMutateRename( |
|||
parent grpc.BidiStreamingServer[filer_pb.StreamMutateEntryRequest, filer_pb.StreamMutateEntryResponse], |
|||
requestId uint64, |
|||
req *filer_pb.StreamRenameEntryRequest, |
|||
) error { |
|||
proxy := &renameStreamProxy{parent: parent, requestId: requestId} |
|||
renameErr := fs.StreamRenameEntry(req, proxy) |
|||
// Always send a final is_last=true to signal rename completion.
|
|||
finalResp := &filer_pb.StreamMutateEntryResponse{ |
|||
RequestId: requestId, |
|||
IsLast: true, |
|||
Response: &filer_pb.StreamMutateEntryResponse_RenameResponse{ |
|||
RenameResponse: &filer_pb.StreamRenameEntryResponse{}, |
|||
}, |
|||
} |
|||
if renameErr != nil { |
|||
finalResp.Error = renameErr.Error() |
|||
finalResp.Errno = renameErrno(renameErr) |
|||
glog.V(0).Infof("StreamMutateEntry rename: %v", renameErr) |
|||
} |
|||
if sendErr := parent.Send(finalResp); sendErr != nil { |
|||
return sendErr |
|||
} |
|||
return nil |
|||
} |
|||
|
|||
// renameStreamProxy adapts the bidi StreamMutateEntry stream to look like a
|
|||
// SeaweedFiler_StreamRenameEntryServer, which is what StreamRenameEntry and
|
|||
// moveEntry expect. Each Send() call forwards the response as a non-final
|
|||
// StreamMutateEntryResponse.
|
|||
type renameStreamProxy struct { |
|||
parent grpc.BidiStreamingServer[filer_pb.StreamMutateEntryRequest, filer_pb.StreamMutateEntryResponse] |
|||
requestId uint64 |
|||
} |
|||
|
|||
func (p *renameStreamProxy) Send(resp *filer_pb.StreamRenameEntryResponse) error { |
|||
return p.parent.Send(&filer_pb.StreamMutateEntryResponse{ |
|||
RequestId: p.requestId, |
|||
IsLast: false, |
|||
Response: &filer_pb.StreamMutateEntryResponse_RenameResponse{RenameResponse: resp}, |
|||
}) |
|||
} |
|||
|
|||
func (p *renameStreamProxy) Context() context.Context { |
|||
return p.parent.Context() |
|||
} |
|||
|
|||
func (p *renameStreamProxy) SendMsg(m any) error { return p.parent.SendMsg(m) } |
|||
func (p *renameStreamProxy) RecvMsg(m any) error { return p.parent.RecvMsg(m) } |
|||
func (p *renameStreamProxy) SetHeader(md metadata.MD) error { return p.parent.SetHeader(md) } |
|||
func (p *renameStreamProxy) SendHeader(md metadata.MD) error { return p.parent.SendHeader(md) } |
|||
func (p *renameStreamProxy) SetTrailer(md metadata.MD) { p.parent.SetTrailer(md) } |
|||
|
|||
// renameErrno maps a rename error to a POSIX errno for the client.
|
|||
func renameErrno(err error) int32 { |
|||
msg := err.Error() |
|||
switch { |
|||
case strings.Contains(msg, "not found"): |
|||
return int32(syscall.ENOENT) |
|||
case strings.Contains(msg, "not empty"): |
|||
return int32(syscall.ENOTEMPTY) |
|||
case strings.Contains(msg, "not directory"): |
|||
return int32(syscall.ENOTDIR) |
|||
default: |
|||
return int32(syscall.EIO) |
|||
} |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue