hive-c0re: per-agent btrfs disk usage + optional quota (#1793)

Follow-up to the btrfs-subvolume migration. Operator-opt-in disk
accounting + quotas on agent state subvolumes via btrfs qgroups:

- three privileged ops (qgroup ops need root): EnsureBtrfsQuota
  (btrfs quota enable on the agent-state filesystem — statfs-gated,
  idempotent, no-op off btrfs), ReadSubvolumeUsage (btrfs qgroup show
  -f --raw for one agent), SetSubvolumeQuota (btrfs qgroup limit, or
  clear). Reuses the is_on_btrfs helper from the subvolume work.
- priv_client wrappers, incl parse_qgroup_usage -> (referenced,
  exclusive) bytes.
- hivectl 'quota' subcommand: enable / show [agent] / limit <agent>
  <size|none>, with a K/M/G/T size parser + human-readable output.

Quota is deliberately NOT auto-enabled: btrfs quota enable triggers a
full rescan that is I/O-heavy on a large filesystem, and the operator
should choose when to pay that. 'quota show' on a plain-dir agent (no
subvolume) reports no qgroup data rather than erroring.

hive-priv/src/main.rs

@@ -155,6 +155,9 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data:
 /// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
 /// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
 /// the returned strings are empty.
+// One match arm per priv op — a flat 1:1 dispatch table. The length tracks
+// the op count, not complexity; splitting it would just scatter the mapping.
+#[allow(clippy::too_many_lines)]
 async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
     match req {
         PrivRequest::StartContainer { ref name } => {
@@ -279,6 +282,21 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
             validate_agent_name(agent_name)?;
             delete_agent_subvolume(agent_name).await
         }
+
+        PrivRequest::EnsureBtrfsQuota => ensure_btrfs_quota().await,
+
+        PrivRequest::ReadSubvolumeUsage { ref agent_name } => {
+            validate_agent_name(agent_name)?;
+            read_subvolume_usage(agent_name).await
+        }
+
+        PrivRequest::SetSubvolumeQuota {
+            ref agent_name,
+            limit_bytes,
+        } => {
+            validate_agent_name(agent_name)?;
+            set_subvolume_quota(agent_name, limit_bytes).await
+        }
     }
 }
 
@@ -621,6 +639,91 @@ async fn delete_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
     Ok((String::new(), String::new()))
 }
 
+/// `EnsureBtrfsQuota` — enable btrfs qgroup accounting on the filesystem
+/// holding `AGENT_STATE_ROOT`. Idempotent + statfs-gated (no-op off btrfs).
+/// Operator opt-in only; see the wire doc.
+async fn ensure_btrfs_quota() -> Result<(String, String)> {
+    let root = PathBuf::from(AGENT_STATE_ROOT);
+    std::fs::create_dir_all(&root)
+        .with_context(|| format!("create agents root {}", root.display()))?;
+    if !is_on_btrfs(&root)? {
+        // Non-btrfs host: quota/qgroups don't apply. No-op success so the
+        // operator-facing verb degrades cleanly.
+        return Ok((
+            String::new(),
+            "filesystem is not btrfs — quota not applicable".to_owned(),
+        ));
+    }
+    let out = Command::new("btrfs")
+        .args(["quota", "enable"])
+        .arg(&root)
+        .output()
+        .await
+        .with_context(|| format!("spawn btrfs quota enable {}", root.display()))?;
+    if !out.status.success() {
+        bail!(
+            "btrfs quota enable {} failed: {}",
+            root.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    tracing::info!(path = %root.display(), "enabled btrfs qgroup accounting");
+    Ok((String::new(), String::new()))
+}
+
+/// `ReadSubvolumeUsage` — return an agent subvolume's qgroup row
+/// (`btrfs qgroup show -f --raw <…/agent_name>`) verbatim in stdout for
+/// hive-c0re to parse. `-f` filters to the subvolume the path lives in, so
+/// the output is the single relevant row. See the wire doc.
+async fn read_subvolume_usage(agent_name: &str) -> Result<(String, String)> {
+    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
+    let out = Command::new("btrfs")
+        .args(["qgroup", "show", "-f", "--raw"])
+        .arg(&agent_root)
+        .output()
+        .await
+        .with_context(|| format!("spawn btrfs qgroup show {}", agent_root.display()))?;
+    if !out.status.success() {
+        // The common failure is "quota not enabled" — pass the stderr
+        // through so hive-c0re can surface it gracefully.
+        bail!(
+            "btrfs qgroup show {} failed: {}",
+            agent_root.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    Ok((
+        String::from_utf8_lossy(&out.stdout).into_owned(),
+        String::new(),
+    ))
+}
+
+/// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent
+/// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
+/// wire doc.
+async fn set_subvolume_quota(
+    agent_name: &str,
+    limit_bytes: Option<u64>,
+) -> Result<(String, String)> {
+    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
+    let limit = limit_bytes.map_or_else(|| "none".to_owned(), |n| n.to_string());
+    let out = Command::new("btrfs")
+        .args(["qgroup", "limit", &limit])
+        .arg(&agent_root)
+        .output()
+        .await
+        .with_context(|| format!("spawn btrfs qgroup limit {}", agent_root.display()))?;
+    if !out.status.success() {
+        bail!(
+            "btrfs qgroup limit {limit} {} failed: {}",
+            agent_root.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    tracing::info!(agent = %agent_name, %limit, "set agent subvolume quota");
+    Ok((String::new(), String::new()))
+}
+
 /// Validate a single argument destined for `forgejo admin`. Rejects
 /// null bytes and newlines (which could corrupt the subprocess args list
 /// or log output). Shell metacharacters are harmless since the command