1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@bokuwalily

Google DriveをAPIなしで記憶の同期先にする ― ファイルIDを壊さない同期設計

1
Posted at

前回、信頼できない外部出力を隔離する話を書きました。今回はその同期経路の話で、gws認証もネットワークコールも一切使わず、Google Drive for Desktopのローカルマウントフォルダを「ただのファイル置き場」として使うmemory-drive-mount-sync.pyの中身です。

困りごと：APIを足さずに同期経路を増やしたい

Vaultの記憶をクラウド側にも置く経路は、すでにmemory-cloud-sync.pyというgws（Google Workspace CLI）ベースの実装があります。OAuth認証、Drive API、Docs APIのbatchUpdate、requiredRevisionIdでのCAS的な競合検知……としっかり作ってありますが、その分依存も重い。

一方でMacにはGoogle Drive for Desktopがすでに動いていて、~/Library/CloudStorage/GoogleDrive-<google-account>/マイドライブ/配下にDriveの中身がローカルファイルとしてマウントされています。ここに書き込めば、認証もAPIコールも足さずに同期できるはずです。

ただし、素朴に「一時ファイルに書いてos.replace()でリネーム」をやると事故ります。os.replace()はinodeを差し替える操作で、Drive for Desktopはローカルのファイル実体（inode）を見てDrive側のfile idと紐付けています。リネームで裏の実体を差し替えると、Drive側からは元のファイルが消えて新しいファイルが生えたように見え、そのファイルが背負っていたfile idを壊してしまいます。

設計：pushは同一ファイルへの上書き、pullは限定的な拾い読み

memory-drive-mount-sync.pyのモジュールdocstringに、このスクリプトの守備範囲がそのまま書かれています。

"""One bounded, single-run local sync against a Google Drive for Desktop mount.

This is deliberately separate from memory-cloud-sync.py: it never calls ``gws``,
never touches network credentials, and does nothing beyond one push (of the
locally built curated snapshot) and one bounded pull (of a handful of plain
dated notes already sitting under the mounted folder). It only ever reads and
writes: the vault's own AI/.runtime build cache, a single named file inside
the mount directory (in place), and AI/INBOX/cloud inside the vault.
"""

役割は3つだけです。

  • push: memory-bridge.py buildでVaultから作ったSHARED-MEMORY.txtを、マウント内の同名ファイルに同一ファイルとして上書きする
  • pull: マウント内にある日付名の平文ノート（.txt/.md/.markdown、最大50件・1件256KB上限）だけを拾い、VaultのAI/INBOX/cloudに隔離ステージングする
  • 単発実行ロックで多重起動を弾く

gwsもDrive APIもないので、requiredRevisionIdのような competing-write 検知はできません。その代わりにPOSIXのfstat/inode一致・O_NOFOLLOW・flockでTOCTOU（check-then-use）を塞ぎます。認証層がない分、ファイルシステム層の几帳面さで安全性を作っています。

push：ファイルIDを壊さない上書き

肝の関数がこれです。os.replace()を使わず、既存のファイルディスクリプタに対してseek(0) → write → truncateします。

def overwrite_shared_memory(path: Path, content: bytes) -> None:
    """Overwrite an existing regular file in place, preserving its identity.

    Deliberately never os.replace(): the mount file's identity (e.g. a Google
    Drive file id behind the local inode) must be preserved rather than
    swapped for a new one.
    """
    initial = m.check_regular_non_symlink(path)
    flags = os.O_RDWR
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    fd = os.open(path, flags)
    try:
        opened = os.fstat(fd)
        if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (initial.st_dev, initial.st_ino):
            raise MountSyncError("mount_target_changed_during_write")
        handle = os.fdopen(fd, "r+b", closefd=False)
        try:
            handle.seek(0)
            handle.write(content)
            handle.truncate()
            handle.flush()
            os.fsync(handle.fileno())
        finally:
            handle.close()
    finally:
        os.close(fd)

check_regular_non_symlinkで最初にlstatベースの安全確認（シンボリックリンクでない・通常ファイルである）をしてからopen()し、直後にfstatした結果の(st_dev, st_ino)を、openする前に見たlstatの結果と突き合わせています。ここが割れていたら、chekcとopenの間でファイルが差し替えられた（TOCTOU）とみなして中断します。

push本体側にも、書く前のガードがあります。

mount_path = mount_dir / "SHARED-MEMORY.txt"
# Missing / symlink / non-regular mount target: refuse to write, surface a
# non-sensitive error code, and never fall back to creating a new file
# (that would mint a new Drive file id in place of the shared one).
m.check_regular_non_symlink(mount_path)

対象ファイルが存在しない・シンボリックリンクになっている・通常ファイルでない場合、新規作成にフォールバックしません。新規作成した瞬間にDrive側で別のfile idが発行されてしまうからです。中身の比較はupdated:行を除いて行い、差分がなければ書き込み自体をskipします（タイムスタンプだけの差分でDrive側の同期イベントを毎回起こさないため）。

pull：拾うのは日付名の平文ノートだけ

pullの対象は明示的に絞ってあります。

MAX_PULL_FILES = 50
MAX_PULL_FILE_BYTES = 256 * 1024
NAME_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}")
PULL_EXTENSIONS = {".txt": "text/plain", ".md": "text/markdown", ".markdown": "text/markdown"}

def classify_pull_candidate(name: str) -> Optional[str]:
    if not NAME_DATE_RE.match(name):
        return None
    if name.upper().startswith("SHARED-MEMORY"):
        return None
    lower = name.lower()
    for suffix, mime in PULL_EXTENSIONS.items():
        if lower.endswith(suffix):
            return mime
    return None

YYYY-MM-DD始まりの名前だけを拾い、push対象そのもの（SHARED-MEMORY始まり）は明示的に除外しています。これがないと、自分がpushしたファイルを次のpullで「未知の外部ノート」として拾い直すループになります。

for entry in entries:
    info = entry.stat(follow_symlinks=False)
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
        counts["skipped"] += 1
        continue
    ...
    candidates.append((entry, mime))
    if len(candidates) >= MAX_PULL_FILES:
        break

シンボリックリンクと非通常ファイルはこの時点で弾き、候補は50件で打ち切ります。フォルダに何千ファイルあっても、走査自体はscandirで軽く、実際に中身を読むのは50件までです。

秘密情報スクリーニングとsymlink安全チェックの使い回し

pullで拾った各ノートは、memory-cloud-sync.pyにすでにあるm.stage_note()にそのまま渡しています。ファイル名がハイフン入りでimport memory-cloud-syncができないため、importlibでパス指定ロードしています。

_CLOUD_SYNC_PATH = Path(__file__).with_name("memory-cloud-sync.py")
_spec = importlib.util.spec_from_file_location("memory_cloud_sync_for_mount_sync", _CLOUD_SYNC_PATH)
if _spec is None or _spec.loader is None:  # pragma: no cover - defensive only
    raise ImportError("memory-cloud-sync.py could not be loaded")
m = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = m
_spec.loader.exec_module(m)

これでm.secure_read（open前後のfstat一致チェック付き読み込み）、m.has_secret/m.is_unsafe（AWSキー・GitHub PAT・sk-系APIキー・JWT形状・Slackトークン・Bearerヘッダ・email・電話番号・医療/告発系ワードなどの正規表現バンク）、m.stage_note（安全なものはAI/INBOX/cloud/<sha256>.mdに provenance 付きで保存、不安全なものは本文を含めない隔離JSONにquarantine/へ）を、一切コードを複製せずに再利用できます。ロジックを2箇所に分散させると、片方だけ直して片方を直し忘れる事故が起きるので、ここは意図的に一元化しています。

多重起動を防ぐ単発実行ロック

launchdが5分おきに起動するので、前回の実行が終わってなければ即skipする必要があります。

lock_path = vault / "AI" / ".runtime" / "memory-drive-mount-sync.lock"
m.ensure_directory_no_symlinks(lock_path.parent)
lock_fd = acquire_lock_fd(lock_path)
...
try:
    fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
    os.close(lock_fd)
    print("locked")
    return 0

acquire_lock_fdはロックファイル自身がシンボリックリンクだったら拒否してからO_NOFOLLOWで開きます。flockが非ブロッキングで取れなければ、エラーではなく"locked"を出してexit 0で正常終了します。多重起動は異常ではなく想定内の事象として扱うのがポイントです。

マウント自体が使えない場合（Driveアプリ未起動・未マウントなど）も区別しています。

def check_mount_dir(path: Path) -> None:
    try:
        info = os.stat(path)
    except OSError as exc:
        raise MountUnavailableError() from exc
    if not stat.S_ISDIR(info.st_mode):
        raise MountUnavailableError()
    if not os.access(path, os.R_OK | os.X_OK):
        raise MountUnavailableError()

これはexit 3のmount_unavailableとして返り、ロック取得失敗の"locked"（exit 0）や本体エラーのexit 2とは別コードにしています。ログを見たときに「まだ動いてる」「マウントが消えてる」「処理中に壊れた」を機械的に切り分けるためです。

launchdの設定：5分おき・低優先度

<key>StartInterval</key>
<integer>300</integer>
<key>LowPriorityIO</key>
<true/>
<key>Nice</key>
<integer>12</integer>
<key>ThrottleInterval</key>
<integer>60</integer>

5分間隔（StartInterval=300）で起動し、LowPriorityIOとNice 12でディスクI/OとCPUの優先度を下げています。同期はネットワークを使わない分、実質ローカルディスクI/Oとmemory-bridge.py buildのCPU時間だけなので、フォアグラウンド作業を邪魔しない優先度で十分です。ThrottleInterval=60はlaunchd側の再起動間隔の下限で、クラッシュループを起こしても1分間隔以上には詰まらせません。

踏んだ落とし穴

  • os.replace()でリネーム上書きするとDrive側のfile idが壊れる → 同一fdに対するseek(0)+write+truncateに変更
  • checkとopenの間でファイルが差し替えられるTOCTOU → lstat結果とfstat結果の(st_dev, st_ino)を突き合わせて不一致なら中断
  • pushしたファイル自身を次のpullで拾ってしまう → classify_pull_candidateでSHARED-MEMORY始まりの名前を明示除外
  • updated:行だけの差分で毎回書き込みが走る → 比較前にupdated:行を取り除いてから同一性判定
  • ファイル名のハイフンでimportできない → importlib.util.spec_from_file_locationでパス指定ロード
  • マウント未使用時に書き込み失敗が"エラー"に埋もれる → mount_unavailableを専用の終了コードに分離

まとめ

  • Google Drive for Desktopのローカルマウントは、認証層を足さずに「ただのファイル置き場」として同期に使える
  • ただしDriveはinodeでfile idを追跡しているため、os.replace()は使わず同一fdへの上書きでファイル実体を保つ
  • 認証層がない分、fstat/inode一致・O_NOFOLLOW・flockというPOSIX側の几帳面さで安全性を代替する
  • pullは日付名・拡張子・件数・サイズで絞り込み、秘密情報スクリーニングとsymlink安全チェックは既存実装から再利用して二重管理を避ける
  • 多重起動・マウント未使用・本体エラーは別々の終了コードで扱い、ログから機械的に切り分けられるようにする

次回は、この単発実行ロックとステージング隔離の考え方を他の同期経路にも広げた話を書く予定です。

Lily（@bokuwalily）― 個人開発者。Claude Code で自動化基盤を組みながら、iOSアプリやWebサービスを量産しています

皆さんの ❤️ やシェアが励みになります！

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?