upgrade: write_atomic era atómico pero NO DURABLE — un corte lo dejaba en 0 bytes

La máquina de estados que promete sobrevivir a un corte dependía de escrituras que no sobreviven a
un corte.

`write_atomic` hacía temporal + `rename`. Eso es atómico frente a OTROS PROCESOS, no frente a un
corte de luz: `rename` sobre un fichero cuyos datos siguen en la caché de página deja, tras el corte,
la entrada nueva apuntando a bloques que nunca se escribieron — un fichero de **CERO BYTES**.

Medido en la caja de producción (SDD 28 §6.14). Apliqué un upgrade y reinicié con
`hcloud server reset`, que es un corte DURO y no un apagado limpio. La caja volvió con:

    pending.json                      0 bytes
    generations/2/manifest.json       0 bytes
    upgrade status   → Error: json: EOF while parsing a value
    upgrade recover  → Error: json: EOF while parsing a value

`recover` existe EXACTAMENTE para «un apply interrumpido por un corte/reinicio» y abortaba con la
huella más probable de ese corte.

Dos arreglos:

1. **`write_atomic` ahora es durable**: `fsync` del temporal ANTES del rename (los datos) y `fsync`
   del DIRECTORIO después (la entrada). Hacen falta los dos; con uno solo sigue habiendo ventana.

2. **Un `pending.json` vacío se reporta como lo que es**: `Error::PendingCorrupt`, que nombra el
   corte, dice que el plan se perdió y apunta al árbol de RESPALDOS, que es lo que sí queda para
   restaurar a mano. Un `json: EOF while parsing a value` crudo manda a mirar el JSON en vez del
   corte.

Tests con su control: sin fichero ⇒ `Ok(None)`; vacío o sólo espacios ⇒ `PendingCorrupt` nombrando
fichero y respaldos; y —el control que hace que valga— un `pending.json` VÁLIDO se sigue leyendo. 23
en verde.

Queda anotado lo que NO se arregló: cuando el manifiesto se pierde, `recover` no puede deshacer solo
(no sabe qué se tocó). El árbol de respaldos tiene la información; reconstruir desde ahí es su propia
unidad de trabajo.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RomoxEGZUhaT4pob1QSX5x

crates/takana-upgrade/src/lib.rs

@@ -65,6 +65,13 @@ pub enum Error {
          `hammer upgrade recover` (completa) o `hammer upgrade recover --rollback` (deshace) antes de seguir"
     )]
     PendingExists { generation: u64 },
+    #[error(
+        "upgrade: el registro del apply pendiente ({path}) está VACÍO. Es la huella de un CORTE a \
+         mitad de escritura: el `rename` llegó al disco y los bytes no. El plan se perdió, así que \
+         no se puede completar ni deshacer automáticamente — pero el árbol de respaldos sigue en \
+         {backups}: restaurá desde ahí, y borrá el `pending.json` y la generación a medias"
+    )]
+    PendingCorrupt { path: String, backups: String },
     #[error("upgrade: {0}")]
     Other(String),
 }
@@ -171,7 +178,17 @@ pub fn pending(state_root: &Path) -> Result<Option<GenerationManifest>> {
     if !p.is_file() {
         return Ok(None);
     }
-    Ok(Some(serde_json::from_slice(&std::fs::read(&p)?)?))
+    let bytes = std::fs::read(&p)?;
+    // Un `pending.json` VACÍO o ilegible es la huella de un corte a mitad de escritura — justo el
+    // caso para el que existe `recover`. Un `EOF while parsing a value` crudo no dice eso: dice que
+    // el JSON está mal, y manda a mirar al lugar equivocado.
+    if bytes.iter().all(|b| b.is_ascii_whitespace()) {
+        return Err(Error::PendingCorrupt {
+            path: p.display().to_string(),
+            backups: generations_dir(state_root).display().to_string(),
+        });
+    }
+    Ok(Some(serde_json::from_slice(&bytes)?))
 }
 
 fn write_pending(state_root: &Path, manifest: &GenerationManifest) -> Result<()> {
@@ -758,14 +775,39 @@ fn tmp_sibling(dst: &Path) -> PathBuf {
     dst.with_file_name(format!(".hammer-upgrade-tmp-{name}"))
 }
 
-/// Escribe `bytes` en `path` de forma atómica (temporal + rename).
+/// Escribe `bytes` en `path` de forma atómica **y DURABLE** (temporal + fsync + rename + fsync del dir).
+///
+/// ⚠ Los dos `fsync` no son prolijidad: sin ellos «atómico» vale sólo frente a OTROS PROCESOS, no
+/// frente a un corte. `rename` sobre un fichero cuyos datos siguen en la caché de página deja, tras
+/// un corte de luz, la entrada nueva apuntando a bloques que nunca se escribieron — o sea un fichero
+/// de **CERO BYTES**.
+///
+/// Medido el 2026-09-11 (SDD 28 §6.14): tras un `hcloud server reset` —que es un corte duro, no un
+/// apagado limpio— la caja volvió con `pending.json` y `generations/2/manifest.json` en 0 bytes, el
+/// upgrade perdido, y `upgrade recover` —cuyo único propósito es sobrevivir exactamente a eso—
+/// abortando con `json: EOF while parsing a value`. La máquina de estados que promete resistir un
+/// corte no puede depender de escrituras que no resisten un corte.
+///
+/// `fsync` del fichero garantiza los datos; `fsync` del DIRECTORIO garantiza que la entrada del
+/// rename sobreviva. Hacen falta los dos.
 fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
     if let Some(parent) = path.parent() {
         std::fs::create_dir_all(parent)?;
     }
     let tmp = tmp_sibling(path);
-    std::fs::write(&tmp, bytes)?;
+    {
+        let mut f = std::fs::File::create(&tmp)?;
+        use std::io::Write;
+        f.write_all(bytes)?;
+        f.sync_all()?; // datos + metadatos del temporal, ANTES del rename
+    }
     std::fs::rename(&tmp, path)?;
+    if let Some(parent) = path.parent() {
+        // La entrada de directorio que el rename creó también necesita llegar al disco.
+        if let Ok(d) = std::fs::File::open(parent) {
+            let _ = d.sync_all();
+        }
+    }
     Ok(())
 }
 
@@ -829,6 +871,55 @@ mod tests {
         h.store_dir_name(name)
     }
 
+    /// Un `pending.json` de CERO BYTES es la huella de un corte a mitad de escritura — el caso para
+    /// el que `recover` existe. Antes reventaba con `json: EOF while parsing a value`, que manda a
+    /// mirar el JSON en vez del corte. Medido en la caja de producción el 2026-09-11 tras un
+    /// `hcloud server reset` (corte duro): `pending.json` y el manifiesto de la generación, los dos
+    /// en 0 bytes, y `upgrade recover` abortando.
+    #[test]
+    fn pending_vacio_se_reporta_como_corte_no_como_json_roto() {
+        let tmp = tempfile::tempdir().unwrap();
+        let state = tmp.path();
+        std::fs::create_dir_all(state).unwrap();
+
+        // (1) sin fichero ⇒ no hay pendiente, sin error
+        assert!(matches!(pending(state), Ok(None)), "sin pending.json no hay intento pendiente");
+
+        // (2) fichero VACÍO ⇒ error de CORTE, nombrando el árbol de respaldos
+        std::fs::write(pending_path(state), b"").unwrap();
+        match pending(state) {
+            Err(Error::PendingCorrupt { path, backups }) => {
+                assert!(path.contains("pending.json"), "el error nombra el fichero: {path}");
+                assert!(!backups.is_empty(), "y dónde están los respaldos");
+            }
+            otro => panic!("esperaba PendingCorrupt, obtuve {otro:?}"),
+        }
+
+        // (3) sólo espacios ⇒ lo mismo (una escritura truncada puede dejar basura en blanco)
+        std::fs::write(pending_path(state), b"  \n").unwrap();
+        assert!(matches!(pending(state), Err(Error::PendingCorrupt { .. })));
+    }
+
+    /// El control que hace que el test de arriba valga: un `pending.json` VÁLIDO se lee igual que
+    /// siempre. Si esto se rompiera, el chequeo nuevo estaría tragándose intentos reales.
+    #[test]
+    fn pending_valido_sigue_leyendose() {
+        let tmp = tempfile::tempdir().unwrap();
+        let state = tmp.path();
+        let m = GenerationManifest {
+            id: 7,
+            tree_dir: "aaaa-x".into(),
+            tree_content: "b3:aaaa".into(),
+            parent: None,
+            created_at: 0,
+            files: vec![],
+            changes: vec![],
+        };
+        write_pending(state, &m).unwrap();
+        let leido = pending(state).unwrap().expect("un pending válido se lee");
+        assert_eq!(leido.id, 7);
+    }
+
     fn h64(prefix: &str) -> String {
         format!("{prefix}{}", "0".repeat(64 - prefix.len()))
     }