michal/tit

Browse tree · Show commit · Download archive

Diff

47c252d2452a → 93ba1e422b59

@@ -1278,6 +1278,7 @@
  "gix-hashtable",
  "gix-object",
  "gix-path",
+ "gix-tempfile",
  "gix-traverse",
  "memmap2",
  "parking_lot",

@@ -14,7 +14,7 @@
 axum = { version = "0.8", default-features = false, features = ["http1", "query", "tokio"] }
 clap = { version = "4.6", default-features = false, features = ["derive", "error-context", "help", "std", "usage"] }
 gix = { version = "0.84", default-features = false, features = ["parallel", "sha1", "sha256"] }
-gix-pack = { version = "0.71", default-features = false, features = ["generate", "sha1", "sha256"] }
+gix-pack = { version = "0.71", default-features = false, features = ["generate", "sha1", "sha256", "streaming-input"] }
 rand = "0.10"
 rusqlite = { version = "0.40", default-features = false, features = ["backup", "bundled"] }
 russh = { version = "0.62", default-features = false, features = ["ring"] }
@@ -22,7 +22,7 @@
 sha2 = { version = "0.11", default-features = false }
 ssh-key = { version = "=0.7.0-rc.11", default-features = false, features = ["ed25519", "p256", "std"] }
 thiserror = "2.0"
-tokio = { version = "1.53", default-features = false, features = ["macros", "net", "rt-multi-thread", "sync", "time"] }
+tokio = { version = "1.53", default-features = false, features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "sync", "time"] }
 toml = { version = "1.1", default-features = false, features = ["parse", "serde", "std"] }
 url = { version = "2.5", features = ["serde"] }
 

@@ -1,0 +1,5 @@
+#!/bin/sh
+set -eu
+
+./scripts/check
+cargo test --locked --release --test git_push_ssh

@@ -1,5 +1,6 @@
 pub(crate) mod http;
 pub(crate) mod packetline;
+pub(crate) mod receive_pack;
 pub(crate) mod repository;
 pub(crate) mod transport;
 pub(crate) mod upload_pack;

@@ -79,6 +79,44 @@
     Ok(packets)
 }
 
+pub(crate) fn first_flush_end(input: &[u8]) -> Result<Option<usize>, PacketLineError> {
+    let mut offset = 0;
+    while offset < input.len() {
+        if offset >= MAX_REQUEST_BYTES {
+            return Err(PacketLineError::RequestTooLarge);
+        }
+        let Some(header) = input.get(offset..offset + 4) else {
+            return Ok(None);
+        };
+        if !header.iter().all(u8::is_ascii_hexdigit) {
+            return Err(PacketLineError::InvalidLength);
+        }
+        let header = std::str::from_utf8(header).map_err(|_| PacketLineError::InvalidLength)?;
+        let length =
+            usize::from_str_radix(header, 16).map_err(|_| PacketLineError::InvalidLength)?;
+        match length {
+            0 if offset + 4 <= MAX_REQUEST_BYTES => return Ok(Some(offset + 4)),
+            0 => return Err(PacketLineError::RequestTooLarge),
+            1 | 2 => offset += 4,
+            3 | 4 => return Err(PacketLineError::InvalidLength),
+            length if length > MAX_PACKET_BYTES => return Err(PacketLineError::PacketTooLarge),
+            length => {
+                let end = offset
+                    .checked_add(length)
+                    .ok_or(PacketLineError::PacketTooLarge)?;
+                if end > input.len() {
+                    return Ok(None);
+                }
+                if end > MAX_REQUEST_BYTES {
+                    return Err(PacketLineError::RequestTooLarge);
+                }
+                offset = end;
+            }
+        }
+    }
+    Ok(None)
+}
+
 #[derive(Debug, Error, Eq, PartialEq)]
 pub(crate) enum PacketLineError {
     #[error("packet-line request is too large")]
@@ -127,6 +165,17 @@
         assert_eq!(
             encode_data(&vec![0; MAX_PACKET_BYTES - 3], &mut Vec::new()),
             Err(PacketLineError::PacketTooLarge)
+        );
+    }
+
+    #[test]
+    fn finds_the_first_complete_flush_boundary() {
+        assert_eq!(first_flush_end(b"0008abc"), Ok(None));
+        assert_eq!(first_flush_end(b"0008abcd0000PACK"), Ok(Some(12)));
+        assert_eq!(first_flush_end(b"0008abcd0001"), Ok(None));
+        assert_eq!(
+            first_flush_end(b"0004"),
+            Err(PacketLineError::InvalidLength)
         );
     }
 }

@@ -1,0 +1,904 @@
+use std::collections::{HashMap, HashSet};
+use std::fs::{self, File};
+use std::io::{BufReader, Write};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use gix::bstr::ByteSlice;
+use gix::hash::{Kind, ObjectId};
+use gix::objs::{CommitRef, Kind as ObjectKind, TagRef, TreeRefIter};
+use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog};
+use gix::refs::{FullName, Target};
+use rand::TryRng;
+use thiserror::Error;
+
+use super::packetline::{Packet, PacketLineError, decode, encode_data, encode_flush};
+use super::upload_pack::hash_name;
+use crate::store::{GitIntentRecord, GitOperationIntent, Store};
+
+const MAX_COMMANDS: usize = 256;
+const MAX_OBJECTS: usize = 100_000;
+const MAX_OBJECT_BYTES: usize = 64 * 1024 * 1024;
+const MAX_PACK_BYTES: u64 = 256 * 1024 * 1024;
+const MAX_WALK_OBJECTS: usize = 500_000;
+const MAX_DELTA_DEPTH: usize = 64;
+const MAX_PROCESSING_TIME: Duration = Duration::from_secs(30);
+
+pub(crate) struct ReceivePack {
+    repository_path: PathBuf,
+    database_path: PathBuf,
+    actor: String,
+    object_format: Kind,
+    intent_id: String,
+    quarantine: PathBuf,
+    incoming_pack: PathBuf,
+    cleanup_on_drop: bool,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+struct RefCommand {
+    old: ObjectId,
+    new: ObjectId,
+    name: FullName,
+}
+
+impl ReceivePack {
+    pub(crate) fn open(
+        repository_path: &Path,
+        database_path: &Path,
+        actor: String,
+    ) -> Result<Self, ReceivePackError> {
+        let repository = open_bare(repository_path)?;
+        let object_format = repository.object_hash();
+        let intent_id = random_id()?;
+        let quarantine = repository_path
+            .join("objects")
+            .join("tit-quarantine")
+            .join(&intent_id);
+        fs::create_dir_all(quarantine.join("pack"))?;
+        let incoming_pack = quarantine.join("incoming.pack");
+        Ok(Self {
+            repository_path: repository_path.to_owned(),
+            database_path: database_path.to_owned(),
+            actor,
+            object_format,
+            intent_id,
+            quarantine,
+            incoming_pack,
+            cleanup_on_drop: true,
+        })
+    }
+
+    pub(crate) fn incoming_pack(&self) -> &Path {
+        &self.incoming_pack
+    }
+
+    pub(crate) fn advertisement(&self) -> Result<Vec<u8>, ReceivePackError> {
+        let repository = open_bare(&self.repository_path)?;
+        let mut references = repository
+            .references()
+            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
+            .all()
+            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
+            .filter_map(|reference| reference.ok())
+            .filter_map(|reference| {
+                let id = reference.try_id()?.detach();
+                Some((reference.name().as_bstr().to_vec(), id))
+            })
+            .collect::<Vec<_>>();
+        references.sort_by(|left, right| left.0.cmp(&right.0));
+        let capabilities = format!(
+            "report-status report-status-v2 delete-refs atomic ofs-delta object-format={} agent=tit/{}",
+            hash_name(self.object_format),
+            env!("CARGO_PKG_VERSION")
+        );
+        let mut output = Vec::new();
+        if references.is_empty() {
+            encode_data(
+                format!(
+                    "{} capabilities^{{}}\0{capabilities}\n",
+                    self.object_format.null()
+                )
+                .as_bytes(),
+                &mut output,
+            )?;
+        } else {
+            for (index, (name, id)) in references.iter().enumerate() {
+                let suffix = if index == 0 {
+                    format!("\0{capabilities}")
+                } else {
+                    String::new()
+                };
+                encode_data(
+                    format!("{id} {}{suffix}\n", String::from_utf8_lossy(name)).as_bytes(),
+                    &mut output,
+                )?;
+            }
+        }
+        encode_flush(&mut output);
+        Ok(output)
+    }
+
+    pub(crate) fn expects_pack(&self, command_bytes: &[u8]) -> Result<bool, ReceivePackError> {
+        Ok(parse_commands(command_bytes, self.object_format)?
+            .iter()
+            .any(|command| !command.new.is_null()))
+    }
+
+    pub(crate) fn finish(&mut self, command_bytes: &[u8]) -> Result<Vec<u8>, ReceivePackError> {
+        let commands = parse_commands(command_bytes, self.object_format)?;
+        let repository = open_bare(&self.repository_path)?;
+        validate_initial_refs(&repository, &commands)?;
+
+        let initial = serialize_refs(&commands, false);
+        let proposed = serialize_refs(&commands, true);
+        let repository_text = path_text(&self.repository_path)?;
+        let quarantine_text = path_text(&self.quarantine)?;
+        let created_at = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| ReceivePackError::Clock)?
+            .as_secs()
+            .try_into()
+            .map_err(|_| ReceivePackError::Clock)?;
+        let mut store = Store::open(&self.database_path)?;
+        store.begin_git_intent(&GitOperationIntent {
+            id: &self.intent_id,
+            repository_path: repository_text,
+            actor: &self.actor,
+            initial_refs: &initial,
+            proposed_refs: &proposed,
+            event_payload: &proposed,
+            quarantine_path: quarantine_text,
+            created_at,
+        })?;
+        self.cleanup_on_drop = false;
+        crash_point("intent");
+
+        let result = (|| {
+            let has_pack =
+                self.incoming_pack.exists() && fs::metadata(&self.incoming_pack)?.len() != 0;
+            let needs_objects = commands.iter().any(|command| !command.new.is_null());
+            let pack_name = if has_pack {
+                self.index_and_validate_pack(&repository, &commands)?
+            } else {
+                if needs_objects {
+                    validate_proposed_objects(&repository, None, &commands)?;
+                }
+                None
+            };
+
+            let pack_name = match pack_name.as_ref() {
+                Some(path) => Some(
+                    path.file_name()
+                        .and_then(|name| name.to_str())
+                        .ok_or(ReceivePackError::Path)?,
+                ),
+                None => None,
+            };
+            store.mark_git_objects_promoted(&self.intent_id, pack_name)?;
+            crash_point("objects");
+            apply_ref_transaction(&repository, &commands)?;
+            crash_point("refs");
+            store.complete_git_intent(&self.intent_id)?;
+            let _ = fs::remove_dir_all(&self.quarantine);
+            crash_point("completed");
+            Ok(status_response(&commands, None)?)
+        })();
+        if result.is_err() {
+            drop(store);
+            recover_incomplete_pushes(&self.database_path)?;
+            self.cleanup_on_drop = true;
+        }
+        result
+    }
+
+    pub(crate) fn rejection_response(
+        &self,
+        command_bytes: &[u8],
+        error: &ReceivePackError,
+    ) -> Vec<u8> {
+        let Ok(commands) = parse_commands(command_bytes, self.object_format) else {
+            return Vec::new();
+        };
+        let mut output = Vec::new();
+        let unpack = if error.is_unpack_error() {
+            format!("unpack {}\n", error.client_reason())
+        } else {
+            "unpack ok\n".to_owned()
+        };
+        if encode_data(unpack.as_bytes(), &mut output).is_err() {
+            return Vec::new();
+        }
+        for command in commands {
+            let name = String::from_utf8_lossy(command.name.as_bstr());
+            let line = format!("ng {name} {}\n", error.client_reason());
+            if encode_data(line.as_bytes(), &mut output).is_err() {
+                return Vec::new();
+            }
+        }
+        encode_flush(&mut output);
+        output
+    }
+
+    fn index_and_validate_pack(
+        &self,
+        repository: &gix::Repository,
+        commands: &[RefCommand],
+    ) -> Result<Option<PathBuf>, ReceivePackError> {
+        let metadata = fs::metadata(&self.incoming_pack)?;
+        if metadata.len() == 0 || metadata.len() > MAX_PACK_BYTES {
+            return Err(ReceivePackError::PackLimit);
+        }
+        let mut reader = BufReader::new(File::open(&self.incoming_pack)?);
+        let mut progress = gix::progress::Discard;
+        let interrupt = Arc::new(AtomicBool::new(false));
+        let timer_interrupt = Arc::clone(&interrupt);
+        let (done_sender, done_receiver) = std::sync::mpsc::channel();
+        let timer = std::thread::spawn(move || {
+            if done_receiver.recv_timeout(MAX_PROCESSING_TIME).is_err() {
+                timer_interrupt.store(true, Ordering::Relaxed);
+            }
+        });
+        let outcome = gix_pack::Bundle::write_to_directory(
+            &mut reader,
+            Some(self.quarantine.join("pack").as_path()),
+            &mut progress,
+            &interrupt,
+            Some(Box::new(repository.objects.clone())),
+            gix_pack::bundle::write::Options {
+                thread_limit: Some(2),
+                iteration_mode: gix_pack::data::input::Mode::Verify,
+                index_version: gix_pack::index::Version::V2,
+                object_hash: self.object_format,
+            },
+        );
+        let _ = done_sender.send(());
+        let _ = timer.join();
+        let outcome = outcome.map_err(|error| {
+            if interrupt.load(Ordering::Relaxed) {
+                ReceivePackError::WallClockLimit
+            } else {
+                ReceivePackError::Pack(error.to_string())
+            }
+        })?;
+        if outcome.index.num_objects > MAX_OBJECTS as u32 {
+            return Err(ReceivePackError::ObjectLimit);
+        }
+        if outcome.index.num_objects == 0 {
+            validate_proposed_objects(repository, None, commands)?;
+            if let Some(path) = outcome.data_path {
+                let _ = fs::remove_file(path);
+            }
+            if let Some(path) = outcome.index_path {
+                let _ = fs::remove_file(path);
+            }
+            if let Some(path) = outcome.keep_path {
+                let _ = fs::remove_file(path);
+            }
+            return Ok(None);
+        }
+        let bundle = outcome
+            .to_bundle()
+            .ok_or(ReceivePackError::MissingPack)?
+            .map_err(|error| ReceivePackError::Pack(error.to_string()))?;
+        validate_delta_depth(&bundle)?;
+        validate_proposed_objects(repository, Some(&bundle), commands)?;
+
+        let source_pack = outcome.data_path.ok_or(ReceivePackError::MissingPack)?;
+        let source_index = outcome.index_path.ok_or(ReceivePackError::MissingPack)?;
+        if fs::metadata(&source_pack)?.len() > MAX_PACK_BYTES {
+            return Err(ReceivePackError::PackLimit);
+        }
+        let destination = self.repository_path.join("objects/pack");
+        fs::create_dir_all(&destination)?;
+        let destination_pack =
+            destination.join(source_pack.file_name().ok_or(ReceivePackError::Path)?);
+        let destination_index =
+            destination.join(source_index.file_name().ok_or(ReceivePackError::Path)?);
+        let pack_exists = destination_pack.exists();
+        let index_exists = destination_index.exists();
+        if pack_exists != index_exists {
+            return Err(ReceivePackError::Repository(
+                "a pack file and its index do not match".to_owned(),
+            ));
+        }
+        let promoted = if pack_exists {
+            fs::remove_file(&source_pack)?;
+            fs::remove_file(&source_index)?;
+            false
+        } else {
+            fs::rename(&source_index, &destination_index)?;
+            if let Err(error) = fs::rename(&source_pack, &destination_pack) {
+                let _ = fs::remove_file(&destination_index);
+                return Err(error.into());
+            }
+            true
+        };
+        if let Some(keep) = outcome.keep_path {
+            let _ = fs::remove_file(keep);
+        }
+        sync_directory(&destination)?;
+        Ok(promoted.then_some(destination_pack))
+    }
+}
+
+fn validate_delta_depth(bundle: &gix_pack::Bundle) -> Result<(), ReceivePackError> {
+    let entries = bundle.index.iter().collect::<Vec<_>>();
+    let offsets_by_id = entries
+        .iter()
+        .map(|entry| (entry.oid, entry.pack_offset))
+        .collect::<HashMap<_, _>>();
+    let headers = entries
+        .iter()
+        .map(|entry| {
+            bundle
+                .pack
+                .entry(entry.pack_offset)
+                .map(|pack_entry| (entry.pack_offset, pack_entry.header))
+                .map_err(|error| ReceivePackError::Pack(error.to_string()))
+        })
+        .collect::<Result<HashMap<_, _>, _>>()?;
+    let mut depths = HashMap::new();
+    let mut visiting = HashSet::new();
+    for offset in headers.keys().copied() {
+        let depth = delta_depth(offset, &headers, &offsets_by_id, &mut depths, &mut visiting)?;
+        if depth > MAX_DELTA_DEPTH {
+            return Err(ReceivePackError::DeltaDepthLimit);
+        }
+    }
+    Ok(())
+}
+
+fn delta_depth(
+    offset: u64,
+    headers: &HashMap<u64, gix_pack::data::entry::Header>,
+    offsets_by_id: &HashMap<ObjectId, u64>,
+    depths: &mut HashMap<u64, usize>,
+    visiting: &mut HashSet<u64>,
+) -> Result<usize, ReceivePackError> {
+    if let Some(depth) = depths.get(&offset) {
+        return Ok(*depth);
+    }
+    if !visiting.insert(offset) {
+        return Err(ReceivePackError::DeltaDepthLimit);
+    }
+    let header = headers.get(&offset).ok_or(ReceivePackError::RecoveryData)?;
+    let depth = match header {
+        gix_pack::data::entry::Header::OfsDelta { base_distance } => {
+            let base = offset
+                .checked_sub(*base_distance)
+                .ok_or(ReceivePackError::DeltaDepthLimit)?;
+            delta_depth(base, headers, offsets_by_id, depths, visiting)? + 1
+        }
+        gix_pack::data::entry::Header::RefDelta { base_id } => {
+            if let Some(base) = offsets_by_id.get(base_id) {
+                delta_depth(*base, headers, offsets_by_id, depths, visiting)? + 1
+            } else {
+                1
+            }
+        }
+        _ => 0,
+    };
+    visiting.remove(&offset);
+    depths.insert(offset, depth);
+    Ok(depth)
+}
+
+impl Drop for ReceivePack {
+    fn drop(&mut self) {
+        if self.cleanup_on_drop {
+            let _ = fs::remove_dir_all(&self.quarantine);
+        }
+    }
+}
+
+pub(crate) fn recover_incomplete_pushes(database_path: &Path) -> Result<(), ReceivePackError> {
+    let mut store = Store::open(database_path)?;
+    for intent in store.incomplete_git_intents()? {
+        recover_intent(&mut store, &intent)?;
+    }
+    Ok(())
+}
+
+fn recover_intent(store: &mut Store, intent: &GitIntentRecord) -> Result<(), ReceivePackError> {
+    let repository_path = Path::new(&intent.repository_path);
+    let repository = open_bare(repository_path)?;
+    let initial = parse_ref_snapshot(&intent.initial_refs, repository.object_hash())?;
+    let proposed = parse_ref_snapshot(&intent.proposed_refs, repository.object_hash())?;
+    let at_initial = refs_match(&repository, &initial)?;
+    let at_proposed = refs_match(&repository, &proposed)?;
+
+    match (intent.state.as_str(), at_initial, at_proposed) {
+        ("pending", true, false) | ("pending", true, true) => {
+            store.abandon_git_intent(&intent.id)?;
+        }
+        ("promoted", false, true) => {
+            store.complete_git_intent(&intent.id)?;
+        }
+        ("promoted", true, false) | ("promoted", true, true) => {
+            remove_promoted_pack(repository_path, intent.pack_name.as_deref())?;
+            store.abandon_git_intent(&intent.id)?;
+        }
+        _ => return Err(ReceivePackError::MixedRecovery(intent.id.clone())),
+    }
+    let quarantine = Path::new(&intent.quarantine_path);
+    let expected_parent = repository_path.join("objects/tit-quarantine");
+    if quarantine.parent() != Some(expected_parent.as_path())
+        || quarantine.file_name().and_then(|name| name.to_str()) != Some(intent.id.as_str())
+    {
+        return Err(ReceivePackError::RecoveryData);
+    }
+    let _ = fs::remove_dir_all(quarantine);
+    Ok(())
+}
+
+fn parse_ref_snapshot(
+    bytes: &[u8],
+    object_format: Kind,
+) -> Result<Vec<(FullName, ObjectId)>, ReceivePackError> {
+    bytes
+        .split(|byte| *byte == b'\n')
+        .filter(|line| !line.is_empty())
+        .map(|line| {
+            let mut fields = line.splitn(2, |byte| *byte == b' ');
+            let id = parse_id(fields.next(), object_format)?;
+            let name = fields.next().ok_or(ReceivePackError::RecoveryData)?;
+            let name =
+                FullName::try_from(name.as_bstr()).map_err(|_| ReceivePackError::RecoveryData)?;
+            Ok((name, id))
+        })
+        .collect()
+}
+
+fn refs_match(
+    repository: &gix::Repository,
+    expected: &[(FullName, ObjectId)],
+) -> Result<bool, ReceivePackError> {
+    for (name, expected_id) in expected {
+        let current = repository
+            .try_find_reference(name)
+            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
+            .and_then(|reference| reference.try_id().map(gix::Id::detach));
+        if (expected_id.is_null() && current.is_some())
+            || (!expected_id.is_null() && current != Some(*expected_id))
+        {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+fn remove_promoted_pack(
+    repository_path: &Path,
+    pack_name: Option<&str>,
+) -> Result<(), ReceivePackError> {
+    let Some(pack_name) = pack_name else {
+        return Ok(());
+    };
+    if Path::new(pack_name)
+        .file_name()
+        .and_then(|name| name.to_str())
+        != Some(pack_name)
+        || !pack_name.starts_with("pack-")
+        || !pack_name.ends_with(".pack")
+    {
+        return Err(ReceivePackError::RecoveryData);
+    }
+    let pack = repository_path.join("objects/pack").join(pack_name);
+    let index = pack.with_extension("idx");
+    if pack.exists() {
+        fs::remove_file(pack)?;
+    }
+    if index.exists() {
+        fs::remove_file(index)?;
+    }
+    sync_directory(&repository_path.join("objects/pack"))?;
+    Ok(())
+}
+
+fn parse_commands(input: &[u8], object_format: Kind) -> Result<Vec<RefCommand>, ReceivePackError> {
+    let packets = decode(input)?;
+    if packets.last() != Some(&Packet::Flush) {
+        return Err(ReceivePackError::Commands);
+    }
+    let mut commands = Vec::new();
+    let mut names = HashSet::new();
+    for (index, packet) in packets[..packets.len() - 1].iter().enumerate() {
+        let Packet::Data(line) = packet else {
+            return Err(ReceivePackError::Commands);
+        };
+        let line = line.strip_suffix(b"\n").unwrap_or(line);
+        let command = if index == 0 {
+            line.split(|byte| *byte == 0).next().unwrap_or_default()
+        } else if line.contains(&0) {
+            return Err(ReceivePackError::Commands);
+        } else {
+            line
+        };
+        let mut fields = command.split(|byte| *byte == b' ');
+        let old = parse_id(fields.next(), object_format)?;
+        let new = parse_id(fields.next(), object_format)?;
+        let name = fields.next().ok_or(ReceivePackError::Commands)?;
+        if fields.next().is_some() || old == new {
+            return Err(ReceivePackError::Commands);
+        }
+        let name = FullName::try_from(name.as_bstr()).map_err(|_| ReceivePackError::RefName)?;
+        if !(name.as_bstr().starts_with(b"refs/heads/")
+            || name.as_bstr().starts_with(b"refs/tags/"))
+            || !names.insert(name.clone())
+        {
+            return Err(ReceivePackError::RefName);
+        }
+        commands.push(RefCommand { old, new, name });
+    }
+    if commands.is_empty() || commands.len() > MAX_COMMANDS {
+        return Err(ReceivePackError::CommandLimit);
+    }
+    Ok(commands)
+}
+
+fn parse_id(input: Option<&[u8]>, object_format: Kind) -> Result<ObjectId, ReceivePackError> {
+    let id = ObjectId::from_hex(input.ok_or(ReceivePackError::Commands)?)
+        .map_err(|_| ReceivePackError::Commands)?;
+    if id.kind() != object_format {
+        return Err(ReceivePackError::ObjectFormat);
+    }
+    Ok(id)
+}
+
+fn validate_initial_refs(
+    repository: &gix::Repository,
+    commands: &[RefCommand],
+) -> Result<(), ReceivePackError> {
+    for command in commands {
+        let current = repository
+            .try_find_reference(&command.name)
+            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
+            .and_then(|reference| reference.try_id().map(gix::Id::detach));
+        match (command.old.is_null(), current) {
+            (true, None) => {}
+            (false, Some(current)) if current == command.old => {}
+            _ => return Err(ReceivePackError::StaleRef),
+        }
+    }
+    Ok(())
+}
+
+fn validate_proposed_objects(
+    repository: &gix::Repository,
+    bundle: Option<&gix_pack::Bundle>,
+    commands: &[RefCommand],
+) -> Result<(), ReceivePackError> {
+    let mut finder = CombinedObjects::new(repository, bundle);
+    for command in commands {
+        if command.new.is_null() {
+            continue;
+        }
+        let kind = finder.kind_and_links(command.new)?.0;
+        if command.name.as_bstr().starts_with(b"refs/heads/") && kind != ObjectKind::Commit {
+            return Err(ReceivePackError::BranchNotCommit);
+        }
+        finder.validate_reachable(command.new)?;
+        if !command.old.is_null()
+            && command.name.as_bstr().starts_with(b"refs/heads/")
+            && !finder.is_ancestor(command.old, command.new)?
+        {
+            return Err(ReceivePackError::NonFastForward);
+        }
+    }
+    Ok(())
+}
+
+struct CombinedObjects<'a> {
+    repository: &'a gix::Repository,
+    bundle: Option<&'a gix_pack::Bundle>,
+}
+
+impl<'a> CombinedObjects<'a> {
+    fn new(repository: &'a gix::Repository, bundle: Option<&'a gix_pack::Bundle>) -> Self {
+        Self { repository, bundle }
+    }
+
+    fn object(&self, id: ObjectId) -> Result<(ObjectKind, Vec<u8>), ReceivePackError> {
+        let mut buffer = Vec::new();
+        let mut inflate = gix::features::zlib::Inflate::default();
+        let mut cache = gix_pack::cache::Never;
+        if let Some(bundle) = self.bundle
+            && let Some((data, _)) = bundle
+                .find(&id, &mut buffer, &mut inflate, &mut cache)
+                .map_err(|error| ReceivePackError::Object(error.to_string()))?
+        {
+            if data.data.len() > MAX_OBJECT_BYTES {
+                return Err(ReceivePackError::ObjectLimit);
+            }
+            return Ok((data.kind, data.data.to_vec()));
+        }
+        let object = self
+            .repository
+            .try_find_object(id)
+            .map_err(|error| ReceivePackError::Object(error.to_string()))?
+            .ok_or(ReceivePackError::MissingObject(id))?;
+        if object.data.len() > MAX_OBJECT_BYTES {
+            return Err(ReceivePackError::ObjectLimit);
+        }
+        Ok((object.kind, object.data.clone()))
+    }
+
+    fn kind_and_links(
+        &self,
+        id: ObjectId,
+    ) -> Result<(ObjectKind, Vec<ObjectId>), ReceivePackError> {
+        let (kind, data) = self.object(id)?;
+        let links = match kind {
+            ObjectKind::Blob => Vec::new(),
+            ObjectKind::Commit => {
+                let commit = CommitRef::from_bytes(&data, id.kind())
+                    .map_err(|_| ReceivePackError::MalformedObject(kind))?;
+                std::iter::once(commit.tree())
+                    .chain(commit.parents())
+                    .collect()
+            }
+            ObjectKind::Tree => TreeRefIter::from_bytes(&data, id.kind())
+                .map(|entry| {
+                    let entry = entry.map_err(|_| ReceivePackError::MalformedObject(kind))?;
+                    Ok((entry.mode.kind() != gix::objs::tree::EntryKind::Commit)
+                        .then(|| entry.oid.to_owned()))
+                })
+                .collect::<Result<Vec<_>, ReceivePackError>>()?
+                .into_iter()
+                .flatten()
+                .collect(),
+            ObjectKind::Tag => vec![
+                TagRef::from_bytes(&data, id.kind())
+                    .map_err(|_| ReceivePackError::MalformedObject(kind))?
+                    .target(),
+            ],
+        };
+        Ok((kind, links))
+    }
+
+    fn validate_reachable(&mut self, root: ObjectId) -> Result<(), ReceivePackError> {
+        let mut pending = vec![root];
+        let mut seen = HashSet::new();
+        while let Some(id) = pending.pop() {
+            if !seen.insert(id) {
+                continue;
+            }
+            if seen.len() > MAX_WALK_OBJECTS {
+                return Err(ReceivePackError::ObjectLimit);
+            }
+            pending.extend(self.kind_and_links(id)?.1);
+        }
+        Ok(())
+    }
+
+    fn is_ancestor(
+        &self,
+        ancestor: ObjectId,
+        descendant: ObjectId,
+    ) -> Result<bool, ReceivePackError> {
+        let mut pending = vec![descendant];
+        let mut seen = HashSet::new();
+        while let Some(id) = pending.pop() {
+            if id == ancestor {
+                return Ok(true);
+            }
+            if !seen.insert(id) {
+                continue;
+            }
+            if seen.len() > MAX_WALK_OBJECTS {
+                return Err(ReceivePackError::ObjectLimit);
+            }
+            let (kind, links) = self.kind_and_links(id)?;
+            if kind != ObjectKind::Commit {
+                return Err(ReceivePackError::BranchNotCommit);
+            }
+            pending.extend(links.into_iter().skip(1));
+        }
+        Ok(false)
+    }
+}
+
+fn apply_ref_transaction(
+    repository: &gix::Repository,
+    commands: &[RefCommand],
+) -> Result<(), ReceivePackError> {
+    let edits = commands.iter().map(|command| RefEdit {
+        name: command.name.clone(),
+        deref: false,
+        change: if command.new.is_null() {
+            Change::Delete {
+                expected: PreviousValue::MustExistAndMatch(Target::Object(command.old)),
+                log: RefLog::AndReference,
+            }
+        } else {
+            Change::Update {
+                expected: if command.old.is_null() {
+                    PreviousValue::MustNotExist
+                } else {
+                    PreviousValue::MustExistAndMatch(Target::Object(command.old))
+                },
+                new: Target::Object(command.new),
+                log: LogChange {
+                    mode: RefLog::AndReference,
+                    force_create_reflog: false,
+                    message: "push".into(),
+                },
+            }
+        },
+    });
+    repository
+        .edit_references_as(edits, None)
+        .map_err(|error| ReceivePackError::RefTransaction(error.to_string()))?;
+    Ok(())
+}
+
+fn status_response(
+    commands: &[RefCommand],
+    error: Option<&str>,
+) -> Result<Vec<u8>, PacketLineError> {
+    let mut output = Vec::new();
+    encode_data(b"unpack ok\n", &mut output)?;
+    for command in commands {
+        let name = String::from_utf8_lossy(command.name.as_bstr());
+        let line = match error {
+            Some(error) => format!("ng {name} {error}\n"),
+            None => format!("ok {name}\n"),
+        };
+        encode_data(line.as_bytes(), &mut output)?;
+    }
+    encode_flush(&mut output);
+    Ok(output)
+}
+
+fn serialize_refs(commands: &[RefCommand], proposed: bool) -> Vec<u8> {
+    let mut output = Vec::new();
+    for command in commands {
+        let id = if proposed { command.new } else { command.old };
+        writeln!(
+            output,
+            "{id} {}",
+            String::from_utf8_lossy(command.name.as_bstr())
+        )
+        .expect("a vector write cannot fail");
+    }
+    output
+}
+
+fn open_bare(path: &Path) -> Result<gix::Repository, ReceivePackError> {
+    let repository =
+        gix::open(path).map_err(|error| ReceivePackError::Repository(error.to_string()))?;
+    if !repository.is_bare() {
+        return Err(ReceivePackError::NotBare);
+    }
+    Ok(repository)
+}
+
+fn random_id() -> Result<String, ReceivePackError> {
+    let mut bytes = [0_u8; 16];
+    rand::rngs::SysRng
+        .try_fill_bytes(&mut bytes)
+        .map_err(|_| ReceivePackError::Random)?;
+    Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect())
+}
+
+fn path_text(path: &Path) -> Result<&str, ReceivePackError> {
+    path.to_str().ok_or(ReceivePackError::Path)
+}
+
+fn sync_directory(path: &Path) -> Result<(), ReceivePackError> {
+    File::open(path)?.sync_all()?;
+    Ok(())
+}
+
+#[cfg(test)]
+fn crash_point(point: &str) {
+    if std::env::var("TIT_M1D_CRASH_AFTER").as_deref() != Ok(point) {
+        return;
+    }
+    let ready = std::env::var_os("TIT_M1D_READY").expect("read the M1D ready path");
+    fs::write(ready, point.as_bytes()).expect("write the M1D ready file");
+    loop {
+        std::thread::park();
+    }
+}
+
+#[cfg(not(test))]
+fn crash_point(_point: &str) {}
+
+#[derive(Debug, Error)]
+pub(crate) enum ReceivePackError {
+    #[error(transparent)]
+    PacketLine(#[from] PacketLineError),
+    #[error(transparent)]
+    Io(#[from] std::io::Error),
+    #[error(transparent)]
+    Store(#[from] crate::store::StoreError),
+    #[error("cannot read Git repository: {0}")]
+    Repository(String),
+    #[error("Git repository is not bare")]
+    NotBare,
+    #[error("receive-pack commands are not valid")]
+    Commands,
+    #[error("receive-pack command count exceeds the limit")]
+    CommandLimit,
+    #[error("receive-pack object ID uses the wrong hash format")]
+    ObjectFormat,
+    #[error("receive-pack reference name is not allowed")]
+    RefName,
+    #[error("receive-pack initial reference does not match")]
+    StaleRef,
+    #[error("receive-pack input is missing a pack")]
+    MissingPack,
+    #[error("receive-pack input has an unexpected pack")]
+    UnexpectedPack,
+    #[error("receive-pack exceeds the pack limit")]
+    PackLimit,
+    #[error("receive-pack exceeds the object limit")]
+    ObjectLimit,
+    #[error("receive-pack exceeds the delta depth limit")]
+    DeltaDepthLimit,
+    #[error("receive-pack exceeds the processing time limit")]
+    WallClockLimit,
+    #[error("cannot index receive-pack input: {0}")]
+    Pack(String),
+    #[error("cannot read received object: {0}")]
+    Object(String),
+    #[error("received object {0} does not exist")]
+    MissingObject(ObjectId),
+    #[error("received {0:?} object is malformed")]
+    MalformedObject(ObjectKind),
+    #[error("a branch target is not a commit")]
+    BranchNotCommit,
+    #[error("a branch update is not a fast-forward")]
+    NonFastForward,
+    #[error("cannot update Git references: {0}")]
+    RefTransaction(String),
+    #[error("incomplete Git operation {0} has mixed reference state")]
+    MixedRecovery(String),
+    #[error("an incomplete Git operation has invalid recovery data")]
+    RecoveryData,
+    #[error("cannot create a random operation ID")]
+    Random,
+    #[error("system clock is before the Unix epoch")]
+    Clock,
+    #[error("filesystem path is not valid UTF-8")]
+    Path,
+}
+
+impl ReceivePackError {
+    fn is_unpack_error(&self) -> bool {
+        matches!(
+            self,
+            Self::MissingPack
+                | Self::UnexpectedPack
+                | Self::PackLimit
+                | Self::ObjectLimit
+                | Self::Pack(_)
+                | Self::Object(_)
+                | Self::MissingObject(_)
+                | Self::MalformedObject(_)
+        )
+    }
+
+    fn client_reason(&self) -> &'static str {
+        match self {
+            Self::StaleRef => "stale reference",
+            Self::NonFastForward => "non-fast-forward",
+            Self::BranchNotCommit => "branch target is not a commit",
+            Self::RefName => "reference name is not allowed",
+            Self::ObjectFormat => "object format does not match",
+            Self::CommandLimit => "too many reference commands",
+            Self::PackLimit => "pack exceeds the byte limit",
+            Self::ObjectLimit => "object limit exceeded",
+            Self::DeltaDepthLimit => "delta depth limit exceeded",
+            Self::WallClockLimit => "processing time limit exceeded",
+            Self::MissingObject(_) => "object connectivity check failed",
+            Self::MixedRecovery(_) => "repository recovery is required",
+            _ if self.is_unpack_error() => "pack validation failed",
+            _ => "push validation failed",
+        }
+    }
+}

@@ -9,6 +9,8 @@
 #[derive(Clone)]
 pub(crate) struct GitRepositories {
     root: PathBuf,
+    push_database: Option<PathBuf>,
+    push_jobs: std::sync::Arc<Semaphore>,
     blocking_jobs: std::sync::Arc<Semaphore>,
 }
 
@@ -20,8 +22,19 @@
         })?;
         Ok(Self {
             root,
+            push_database: None,
+            push_jobs: std::sync::Arc::new(Semaphore::new(1)),
             blocking_jobs: std::sync::Arc::new(Semaphore::new(MAX_BLOCKING_GIT_JOBS)),
         })
+    }
+
+    pub(crate) fn new_with_pushes(
+        root: &Path,
+        database: &Path,
+    ) -> Result<Self, RepositoryPathError> {
+        let mut repositories = Self::new(root)?;
+        repositories.push_database = Some(database.to_owned());
+        Ok(repositories)
     }
 
     pub(crate) fn resolve(
@@ -75,9 +88,52 @@
         self.resolve(owner, repository)
     }
 
+    pub(crate) fn resolve_ssh_service(
+        &self,
+        command: &[u8],
+    ) -> Result<GitSshService, RepositoryPathError> {
+        if command.starts_with(b"git-upload-pack ") {
+            return self.resolve_ssh_command(command).map(GitSshService::Upload);
+        }
+        let command =
+            std::str::from_utf8(command).map_err(|_| RepositoryPathError::InvalidCommand)?;
+        let path = command
+            .strip_prefix("git-receive-pack '")
+            .and_then(|value| value.strip_suffix('\''))
+            .ok_or(RepositoryPathError::InvalidCommand)?;
+        if path.contains('\'') {
+            return Err(RepositoryPathError::InvalidCommand);
+        }
+        let path = path.strip_prefix('/').unwrap_or(path);
+        let mut components = path.split('/');
+        let owner = components
+            .next()
+            .ok_or(RepositoryPathError::InvalidCommand)?;
+        let repository = components
+            .next()
+            .ok_or(RepositoryPathError::InvalidCommand)?;
+        if components.next().is_some() {
+            return Err(RepositoryPathError::InvalidCommand);
+        }
+        self.resolve(owner, repository).map(GitSshService::Receive)
+    }
+
+    pub(crate) fn push_database(&self) -> Option<&Path> {
+        self.push_database.as_deref()
+    }
+
+    pub(crate) async fn push_permit(&self) -> Result<OwnedSemaphorePermit, AcquireError> {
+        self.push_jobs.clone().acquire_owned().await
+    }
+
     pub(crate) async fn blocking_permit(&self) -> Result<OwnedSemaphorePermit, AcquireError> {
         self.blocking_jobs.clone().acquire_owned().await
     }
+}
+
+pub(crate) enum GitSshService {
+    Upload(PathBuf),
+    Receive(PathBuf),
 }
 
 fn valid_name(value: &str, maximum: usize) -> bool {
@@ -134,6 +190,12 @@
                 .expect("resolve a repository"),
             fs::canonicalize(&repository).expect("canonicalize the repository")
         );
+        assert!(matches!(
+            repositories
+                .resolve_ssh_service(b"git-receive-pack '/alice/example.git'")
+                .expect("resolve receive-pack"),
+            GitSshService::Receive(path) if path == fs::canonicalize(&repository).expect("canonicalize the repository")
+        ));
         assert_eq!(
             repositories
                 .resolve_ssh_command(b"git-upload-pack '/alice/example.git'")

@@ -310,7 +310,7 @@
     line.strip_suffix(b"\n").unwrap_or(line)
 }
 
-fn hash_name(kind: Kind) -> &'static str {
+pub(crate) fn hash_name(kind: Kind) -> &'static str {
     match kind {
         Kind::Sha1 => "sha1",
         Kind::Sha256 => "sha256",

@@ -10,17 +10,20 @@
 use russh::{Channel, ChannelId, MethodKind, MethodSet, Preferred, Pty};
 use ssh_key::{Algorithm, EcdsaCurve, PrivateKey, PublicKey};
 use thiserror::Error;
+use tokio::io::AsyncWriteExt;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 
 use crate::auth::SshPublicKey;
-use crate::git::packetline::{MAX_REQUEST_BYTES, Packet, decode, encode_data};
-use crate::git::transport::GitRepositories;
+use crate::git::packetline::{MAX_REQUEST_BYTES, Packet, decode, encode_data, first_flush_end};
+use crate::git::receive_pack::{ReceivePack, ReceivePackError};
+use crate::git::transport::{GitRepositories, GitSshService};
 use crate::git::upload_pack::{ProtocolVersion, UploadPack, UploadPackError};
 
 const VERSION_COMMAND: &[u8] = b"tit --version";
 const GIT_PROTOCOL_VARIABLE: &str = "GIT_PROTOCOL";
+const MAX_RECEIVE_PACK_BYTES: u64 = 128 * 1024 * 1024;
 
 pub(crate) struct RunningSshServer {
     address: SocketAddr,
@@ -34,7 +37,7 @@
         address: SocketAddr,
         authorized_keys: &[SshPublicKey],
     ) -> Result<Self, SshServerError> {
-        Self::start_inner(address, authorized_keys, None).await
+        Self::start_inner(address, authorized_keys, &[], None).await
     }
 
     pub(crate) async fn start_with_git(
@@ -42,12 +45,32 @@
         authorized_keys: &[SshPublicKey],
         repositories: GitRepositories,
     ) -> Result<Self, SshServerError> {
-        Self::start_inner(address, authorized_keys, Some(repositories)).await
+        Self::start_inner(address, authorized_keys, &[], Some(repositories)).await
+    }
+
+    pub(crate) async fn start_with_git_writes(
+        address: SocketAddr,
+        authorized_keys: &[SshPublicKey],
+        writable_keys: &[SshPublicKey],
+        repositories: GitRepositories,
+    ) -> Result<Self, SshServerError> {
+        let database = repositories
+            .push_database()
+            .ok_or_else(|| SshServerError::Recovery("push storage is not configured".to_owned()))?
+            .to_owned();
+        tokio::task::spawn_blocking(move || {
+            crate::git::receive_pack::recover_incomplete_pushes(&database)
+        })
+        .await
+        .map_err(|_| SshServerError::Join)?
+        .map_err(|error| SshServerError::Recovery(error.to_string()))?;
+        Self::start_inner(address, authorized_keys, writable_keys, Some(repositories)).await
     }
 
     async fn start_inner(
         address: SocketAddr,
         authorized_keys: &[SshPublicKey],
+        writable_keys: &[SshPublicKey],
         repositories: Option<GitRepositories>,
     ) -> Result<Self, SshServerError> {
         let listener = TcpListener::bind(address).await?;
@@ -74,8 +97,14 @@
             nodelay: true,
             ..Default::default()
         });
-        let authorized_keys = Arc::new(
+        let authorized_keys: Arc<HashMap<PublicKey, String>> = Arc::new(
             authorized_keys
+                .iter()
+                .map(|key| (key.public_key().clone(), key.fingerprint().to_owned()))
+                .collect(),
+        );
+        let writable_keys = Arc::new(
+            writable_keys
                 .iter()
                 .map(|key| key.public_key().clone())
                 .collect(),
@@ -83,6 +112,7 @@
         let audit = Arc::new(RequestAudit::default());
         let server = SshServer {
             authorized_keys,
+            writable_keys,
             audit: Arc::clone(&audit),
             repositories: repositories.map(Arc::new),
         };
@@ -127,11 +157,14 @@
     Startup,
     #[error("SSH server task failed")]
     Join,
+    #[error("cannot recover Git writes: {0}")]
+    Recovery(String),
 }
 
 #[derive(Clone)]
 struct SshServer {
-    authorized_keys: Arc<HashSet<PublicKey>>,
+    authorized_keys: Arc<HashMap<PublicKey, String>>,
+    writable_keys: Arc<HashSet<PublicKey>>,
     audit: Arc<RequestAudit>,
     repositories: Option<Arc<GitRepositories>>,
 }
@@ -142,25 +175,44 @@
     fn new_client(&mut self, _peer_address: Option<SocketAddr>) -> Self::Handler {
         SshSession {
             authorized_keys: Arc::clone(&self.authorized_keys),
+            writable_keys: Arc::clone(&self.writable_keys),
             audit: Arc::clone(&self.audit),
             repositories: self.repositories.clone(),
             protocol: ProtocolVersion::V0,
             git_channels: HashMap::new(),
+            authenticated_actor: None,
+            authenticated_writer: false,
         }
     }
 }
 
 struct SshSession {
-    authorized_keys: Arc<HashSet<PublicKey>>,
+    authorized_keys: Arc<HashMap<PublicKey, String>>,
+    writable_keys: Arc<HashSet<PublicKey>>,
     audit: Arc<RequestAudit>,
     repositories: Option<Arc<GitRepositories>>,
     protocol: ProtocolVersion,
     git_channels: HashMap<ChannelId, GitChannel>,
+    authenticated_actor: Option<String>,
+    authenticated_writer: bool,
 }
 
-struct GitChannel {
+enum GitChannel {
+    Upload(Box<UploadChannel>),
+    Receive(Box<ReceiveChannel>),
+}
+
+struct UploadChannel {
     service: UploadPack,
     request: Vec<u8>,
+}
+
+struct ReceiveChannel {
+    service: ReceivePack,
+    commands: Vec<u8>,
+    commands_complete: bool,
+    pack: tokio::fs::File,
+    pack_bytes: u64,
 }
 
 impl Handler for SshSession {
@@ -179,7 +231,13 @@
         _user: &str,
         public_key: &PublicKey,
     ) -> Result<Auth, Self::Error> {
-        Ok(self.authorize(public_key))
+        if let Some(actor) = self.authorized_keys.get(public_key) {
+            self.authenticated_actor = Some(actor.clone());
+            self.authenticated_writer = self.writable_keys.contains(public_key);
+            Ok(Auth::Accept)
+        } else {
+            Ok(Auth::reject())
+        }
     }
 
     async fn channel_open_session(
@@ -310,34 +368,47 @@
             session.eof(channel)?;
             session.close(channel)?;
         } else {
-            let service = if let Some(repositories) = &self.repositories
-                && let Ok(path) = repositories.resolve_ssh_command(command)
-                && let Ok(permit) = repositories.blocking_permit().await
-            {
-                let protocol = self.protocol;
-                tokio::task::spawn_blocking(move || {
-                    let _permit = permit;
-                    let service = UploadPack::open(&path)?;
-                    let advertisement = service.advertisement(protocol, false)?;
-                    Ok::<_, UploadPackError>((service, advertisement))
-                })
-                .await
-                .ok()
-                .and_then(Result::ok)
-            } else {
-                None
-            };
-            if let Some((service, advertisement)) = service {
+            let service = self.open_git_service(command).await;
+            if let Some(service) = service {
                 self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
                 session.channel_success(channel)?;
-                session.data(channel, advertisement)?;
-                self.git_channels.insert(
-                    channel,
-                    GitChannel {
+                match service {
+                    InitialGitService::Upload {
                         service,
-                        request: Vec::new(),
-                    },
-                );
+                        advertisement,
+                    } => {
+                        session.data(channel, advertisement)?;
+                        self.git_channels.insert(
+                            channel,
+                            GitChannel::Upload(Box::new(UploadChannel {
+                                service: *service,
+                                request: Vec::new(),
+                            })),
+                        );
+                    }
+                    InitialGitService::Receive {
+                        service,
+                        advertisement,
+                    } => {
+                        session.data(channel, advertisement)?;
+                        let pack = tokio::fs::File::create(service.incoming_pack()).await;
+                        match pack {
+                            Ok(pack) => {
+                                self.git_channels.insert(
+                                    channel,
+                                    GitChannel::Receive(Box::new(ReceiveChannel {
+                                        service,
+                                        commands: Vec::new(),
+                                        commands_complete: false,
+                                        pack,
+                                        pack_bytes: 0,
+                                    })),
+                                );
+                            }
+                            Err(_) => fail_git_channel(channel, session)?,
+                        }
+                    }
+                }
             } else {
                 self.audit.rejected_exec.fetch_add(1, Ordering::Relaxed);
                 session.channel_failure(channel)?;
@@ -353,8 +424,27 @@
         data: &[u8],
         session: &mut Session,
     ) -> Result<(), Self::Error> {
-        let Some(mut git) = self.git_channels.remove(&channel) else {
+        let Some(git) = self.git_channels.remove(&channel) else {
             return Ok(());
+        };
+        let mut git = match git {
+            GitChannel::Upload(git) => git,
+            GitChannel::Receive(mut git) => {
+                if receive_data(&mut git, data).await.is_err() {
+                    fail_git_channel(channel, session)?;
+                } else if git.commands_complete
+                    && matches!(git.service.expects_pack(&git.commands), Ok(false))
+                {
+                    send_receive_result(
+                        channel,
+                        finish_receive(self.repositories.clone(), git).await,
+                        session,
+                    )?;
+                } else {
+                    self.git_channels.insert(channel, GitChannel::Receive(git));
+                }
+                return Ok(());
+            }
         };
         if git.request.len().saturating_add(data.len()) > MAX_REQUEST_BYTES {
             fail_git_channel(channel, session)?;
@@ -366,7 +456,7 @@
             Ok(packets) => packets,
             Err(super::git::packetline::PacketLineError::TruncatedHeader)
             | Err(super::git::packetline::PacketLineError::TruncatedPacket) => {
-                self.git_channels.insert(channel, git);
+                self.git_channels.insert(channel, GitChannel::Upload(git));
                 return Ok(());
             }
             Err(_) => {
@@ -393,12 +483,12 @@
                         Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
                     }
                 } else {
-                    self.git_channels.insert(channel, git);
+                    self.git_channels.insert(channel, GitChannel::Upload(git));
                 }
             }
             ProtocolVersion::V2 => {
                 if packets.last() != Some(&Packet::Flush) {
-                    self.git_channels.insert(channel, git);
+                    self.git_channels.insert(channel, GitChannel::Upload(git));
                     return Ok(());
                 }
                 let fetch = packets.iter().any(
@@ -414,7 +504,7 @@
                             finish_git_channel(channel, 0, session)?;
                         } else {
                             git.request.clear();
-                            self.git_channels.insert(channel, git);
+                            self.git_channels.insert(channel, GitChannel::Upload(git));
                         }
                     }
                     Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
@@ -430,6 +520,23 @@
         _session: &mut Session,
     ) -> Result<(), Self::Error> {
         self.git_channels.remove(&channel);
+        Ok(())
+    }
+
+    async fn channel_eof(
+        &mut self,
+        channel: ChannelId,
+        session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        let Some(GitChannel::Receive(git)) = self.git_channels.remove(&channel) else {
+            return Ok(());
+        };
+        if !git.commands_complete {
+            fail_git_channel(channel, session)?;
+            return Ok(());
+        }
+        let result = finish_receive(self.repositories.clone(), git).await;
+        send_receive_result(channel, result, session)?;
         Ok(())
     }
 
@@ -477,19 +584,166 @@
 
 impl SshSession {
     fn authorize(&self, public_key: &PublicKey) -> Auth {
-        if self.authorized_keys.contains(public_key) {
+        if self.authorized_keys.contains_key(public_key) {
             Auth::Accept
         } else {
             Auth::reject()
         }
     }
+
+    async fn open_git_service(&mut self, command: &[u8]) -> Option<InitialGitService> {
+        let repositories = self.repositories.as_ref()?;
+        let service = repositories.resolve_ssh_service(command).ok()?;
+        match service {
+            GitSshService::Upload(path) => {
+                let permit = repositories.blocking_permit().await.ok()?;
+                let protocol = self.protocol;
+                tokio::task::spawn_blocking(move || {
+                    let _permit = permit;
+                    let service = UploadPack::open(&path)?;
+                    let advertisement = service.advertisement(protocol, false)?;
+                    Ok::<_, UploadPackError>(InitialGitService::Upload {
+                        service: Box::new(service),
+                        advertisement,
+                    })
+                })
+                .await
+                .ok()?
+                .ok()
+            }
+            GitSshService::Receive(path) => {
+                if !self.authenticated_writer {
+                    return None;
+                }
+                let database = repositories.push_database()?.to_owned();
+                let actor = self.authenticated_actor.clone()?;
+                let permit = repositories.blocking_permit().await.ok()?;
+                tokio::task::spawn_blocking(move || {
+                    let _permit = permit;
+                    let service = ReceivePack::open(&path, &database, actor)?;
+                    let advertisement = service.advertisement()?;
+                    Ok::<_, ReceivePackError>(InitialGitService::Receive {
+                        service,
+                        advertisement,
+                    })
+                })
+                .await
+                .ok()?
+                .ok()
+            }
+        }
+    }
+}
+
+enum InitialGitService {
+    Upload {
+        service: Box<UploadPack>,
+        advertisement: Vec<u8>,
+    },
+    Receive {
+        service: ReceivePack,
+        advertisement: Vec<u8>,
+    },
+}
+
+async fn receive_data(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
+    if git.commands_complete {
+        write_receive_pack(git, data).await?;
+        return Ok(());
+    }
+    git.commands.extend_from_slice(data);
+    let boundary = first_flush_end(&git.commands).map_err(|_| ())?;
+    let Some(boundary) = boundary else {
+        return Ok(());
+    };
+    let pack = git.commands.split_off(boundary);
+    git.commands_complete = true;
+    write_receive_pack(git, &pack).await
+}
+
+async fn write_receive_pack(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
+    let bytes = u64::try_from(data.len()).map_err(|_| ())?;
+    git.pack_bytes = git.pack_bytes.checked_add(bytes).ok_or(())?;
+    if git.pack_bytes > MAX_RECEIVE_PACK_BYTES {
+        return Err(());
+    }
+    git.pack.write_all(data).await.map_err(|_| ())
+}
+
+type ReceiveResult = Option<Result<Vec<u8>, (ReceivePackError, Vec<u8>)>>;
+
+async fn finish_receive(
+    repositories: Option<Arc<GitRepositories>>,
+    git: Box<ReceiveChannel>,
+) -> ReceiveResult {
+    let ReceiveChannel {
+        mut service,
+        commands,
+        mut pack,
+        ..
+    } = *git;
+    pack.flush().await.ok()?;
+    pack.sync_all().await.ok()?;
+    drop(pack);
+    let repositories = repositories?;
+    let push_permit = repositories.push_permit().await.ok()?;
+    let permit = repositories.blocking_permit().await.ok()?;
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        let _push_permit = push_permit;
+        match service.finish(&commands) {
+            Ok(response) => Ok(response),
+            Err(error) => {
+                let response = service.rejection_response(&commands, &error);
+                Err((error, response))
+            }
+        }
+    })
+    .await
+    .ok()
+}
+
+fn send_receive_result(
+    channel: ChannelId,
+    result: ReceiveResult,
+    session: &mut Session,
+) -> Result<(), russh::Error> {
+    match result {
+        Some(Ok(response)) => {
+            session.data(channel, response)?;
+            finish_git_channel(channel, 0, session)
+        }
+        Some(Err((error, response))) => {
+            eprintln!("tit: receive-pack failed: {error}");
+            session.data(
+                channel,
+                if response.is_empty() {
+                    receive_error_response()
+                } else {
+                    response
+                },
+            )?;
+            finish_git_channel(channel, 1, session)
+        }
+        None => {
+            session.data(channel, receive_error_response())?;
+            finish_git_channel(channel, 1, session)
+        }
+    }
+}
+
+fn receive_error_response() -> Vec<u8> {
+    let mut response = Vec::new();
+    let _ = encode_data(b"unpack tit rejected the push\n", &mut response);
+    super::git::packetline::encode_flush(&mut response);
+    response
 }
 
 async fn respond_git(
     repositories: Option<Arc<GitRepositories>>,
     protocol: ProtocolVersion,
-    git: GitChannel,
-) -> Option<(GitChannel, Result<Vec<u8>, UploadPackError>)> {
+    git: Box<UploadChannel>,
+) -> Option<(Box<UploadChannel>, Result<Vec<u8>, UploadPackError>)> {
     let permit = repositories?.blocking_permit().await.ok()?;
     tokio::task::spawn_blocking(move || {
         let _permit = permit;

@@ -1,0 +1,23 @@
+CREATE TABLE git_operation_intent (
+    id TEXT PRIMARY KEY CHECK (length(id) = 32),
+    repository_path TEXT NOT NULL CHECK (length(repository_path) BETWEEN 1 AND 4096),
+    actor TEXT NOT NULL CHECK (length(actor) BETWEEN 1 AND 256),
+    initial_refs BLOB NOT NULL,
+    proposed_refs BLOB NOT NULL,
+    event_payload BLOB NOT NULL,
+    quarantine_path TEXT NOT NULL CHECK (length(quarantine_path) BETWEEN 1 AND 4096),
+    state TEXT NOT NULL CHECK (state IN ('pending', 'promoted', 'completed', 'abandoned')),
+    pack_name TEXT,
+    created_at INTEGER NOT NULL CHECK (created_at >= 0)
+) STRICT;
+
+CREATE INDEX git_operation_intent_incomplete
+ON git_operation_intent (state, created_at, id)
+WHERE state IN ('pending', 'promoted');
+
+CREATE TABLE git_operation_event (
+    id INTEGER PRIMARY KEY,
+    intent_id TEXT NOT NULL UNIQUE
+        REFERENCES git_operation_intent (id) ON DELETE RESTRICT,
+    payload BLOB NOT NULL
+) STRICT;

@@ -9,7 +9,7 @@
 
 const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
 const BUSY_TIMEOUT_MILLISECONDS: i64 = 5_000;
-const SCHEMA_VERSION: i64 = 2;
+const SCHEMA_VERSION: i64 = 3;
 #[allow(
     dead_code,
     reason = "the integration test imports this module without the CLI operation"
@@ -19,9 +19,10 @@
     dead_code,
     reason = "M1A proves migrations before the M2 server calls them"
 )]
-const MIGRATIONS: [&str; 2] = [
+const MIGRATIONS: [&str; 3] = [
     include_str!("migrations/001_initial.sql"),
     include_str!("migrations/002_state.sql"),
+    include_str!("migrations/003_git_intents.sql"),
 ];
 
 #[derive(Debug, Error)]
@@ -48,6 +49,8 @@
         expected: &'static str,
         actual: String,
     },
+    #[error("Git operation intent {0} is not in the required state")]
+    IntentState(String),
 }
 
 pub(crate) struct Store {
@@ -181,6 +184,129 @@
     )]
     pub(crate) fn connection_mut(&mut self) -> &mut Connection {
         &mut self.connection
+    }
+
+    pub(crate) fn begin_git_intent(
+        &self,
+        intent: &GitOperationIntent<'_>,
+    ) -> Result<(), StoreError> {
+        self.connection.execute(
+            "INSERT INTO git_operation_intent
+             (id, repository_path, actor, initial_refs, proposed_refs, event_payload,
+              quarantine_path, state, created_at)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'pending', ?8)",
+            rusqlite::params![
+                intent.id,
+                intent.repository_path,
+                intent.actor,
+                intent.initial_refs,
+                intent.proposed_refs,
+                intent.event_payload,
+                intent.quarantine_path,
+                intent.created_at,
+            ],
+        )?;
+        Ok(())
+    }
+
+    pub(crate) fn mark_git_objects_promoted(
+        &self,
+        id: &str,
+        pack_name: Option<&str>,
+    ) -> Result<(), StoreError> {
+        let changed = self.connection.execute(
+            "UPDATE git_operation_intent
+             SET state = 'promoted', pack_name = ?2
+             WHERE id = ?1 AND state = 'pending'",
+            rusqlite::params![id, pack_name],
+        )?;
+        require_one_intent(id, changed)
+    }
+
+    pub(crate) fn complete_git_intent(&mut self, id: &str) -> Result<(), StoreError> {
+        let transaction = self
+            .connection
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let payload: Vec<u8> = transaction.query_row(
+            "SELECT event_payload FROM git_operation_intent
+             WHERE id = ?1 AND state = 'promoted'",
+            [id],
+            |row| row.get(0),
+        )?;
+        let changed = transaction.execute(
+            "UPDATE git_operation_intent SET state = 'completed'
+             WHERE id = ?1 AND state = 'promoted'",
+            [id],
+        )?;
+        require_one_intent(id, changed)?;
+        transaction.execute(
+            "INSERT INTO git_operation_event (intent_id, payload) VALUES (?1, ?2)",
+            rusqlite::params![id, payload],
+        )?;
+        transaction.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn abandon_git_intent(&self, id: &str) -> Result<(), StoreError> {
+        let changed = self.connection.execute(
+            "UPDATE git_operation_intent SET state = 'abandoned'
+             WHERE id = ?1 AND state IN ('pending', 'promoted')",
+            [id],
+        )?;
+        require_one_intent(id, changed)
+    }
+
+    pub(crate) fn incomplete_git_intents(&self) -> Result<Vec<GitIntentRecord>, StoreError> {
+        let mut statement = self.connection.prepare(
+            "SELECT id, repository_path, initial_refs, proposed_refs, quarantine_path,
+                    state, pack_name
+             FROM git_operation_intent
+             WHERE state IN ('pending', 'promoted')
+             ORDER BY created_at, id",
+        )?;
+        let records = statement
+            .query_map([], |row| {
+                Ok(GitIntentRecord {
+                    id: row.get(0)?,
+                    repository_path: row.get(1)?,
+                    initial_refs: row.get(2)?,
+                    proposed_refs: row.get(3)?,
+                    quarantine_path: row.get(4)?,
+                    state: row.get(5)?,
+                    pack_name: row.get(6)?,
+                })
+            })?
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(records)
+    }
+}
+
+pub(crate) struct GitOperationIntent<'a> {
+    pub(crate) id: &'a str,
+    pub(crate) repository_path: &'a str,
+    pub(crate) actor: &'a str,
+    pub(crate) initial_refs: &'a [u8],
+    pub(crate) proposed_refs: &'a [u8],
+    pub(crate) event_payload: &'a [u8],
+    pub(crate) quarantine_path: &'a str,
+    pub(crate) created_at: i64,
+}
+
+pub(crate) struct GitIntentRecord {
+    pub(crate) id: String,
+    pub(crate) repository_path: String,
+    pub(crate) initial_refs: Vec<u8>,
+    pub(crate) proposed_refs: Vec<u8>,
+    pub(crate) quarantine_path: String,
+    pub(crate) state: String,
+    pub(crate) pack_name: Option<String>,
+}
+
+fn require_one_intent(id: &str, changed: usize) -> Result<(), StoreError> {
+    if changed == 1 {
+        Ok(())
+    } else {
+        Err(StoreError::IntentState(id.to_owned()))
     }
 }
 

@@ -9,7 +9,7 @@
 };
 use tempfile::TempDir;
 
-const V2_DATABASE: &str = include_str!("fixtures/sqlite/v2.sql");
+const V3_DATABASE: &str = include_str!("fixtures/sqlite/v3.sql");
 
 #[test]
 fn help_and_version_use_standard_output() {
@@ -64,7 +64,7 @@
     let database = rusqlite::Connection::open(instance.path().join("tit.sqlite3"))
         .expect("open the instance database");
     database
-        .execute_batch(V2_DATABASE)
+        .execute_batch(V3_DATABASE)
         .expect("create the current database");
     drop(database);
 
@@ -116,7 +116,7 @@
     let database = rusqlite::Connection::open(instance.path().join("tit.sqlite3"))
         .expect("open the instance database");
     database
-        .execute_batch(V2_DATABASE)
+        .execute_batch(V3_DATABASE)
         .expect("create the current database");
     database
         .pragma_update(None, "foreign_keys", false)

@@ -1,0 +1,51 @@
+CREATE TABLE m1a_parent (
+    id INTEGER PRIMARY KEY,
+    name TEXT NOT NULL UNIQUE CHECK (length(name) BETWEEN 1 AND 64),
+    created_at INTEGER NOT NULL CHECK (created_at >= 0)
+) STRICT;
+
+CREATE INDEX m1a_parent_created_at ON m1a_parent (created_at, id);
+
+CREATE TABLE m1a_child (
+    id INTEGER PRIMARY KEY,
+    parent_id INTEGER NOT NULL REFERENCES m1a_parent (id) ON DELETE RESTRICT,
+    sequence INTEGER NOT NULL CHECK (sequence >= 0),
+    body TEXT NOT NULL,
+    state TEXT NOT NULL DEFAULT 'open' CHECK (state IN ('open', 'closed')),
+    UNIQUE (parent_id, sequence)
+) STRICT;
+
+CREATE INDEX m1a_child_state_parent
+ON m1a_child (state, parent_id, sequence);
+
+CREATE TABLE git_operation_intent (
+    id TEXT PRIMARY KEY CHECK (length(id) = 32),
+    repository_path TEXT NOT NULL CHECK (length(repository_path) BETWEEN 1 AND 4096),
+    actor TEXT NOT NULL CHECK (length(actor) BETWEEN 1 AND 256),
+    initial_refs BLOB NOT NULL,
+    proposed_refs BLOB NOT NULL,
+    event_payload BLOB NOT NULL,
+    quarantine_path TEXT NOT NULL CHECK (length(quarantine_path) BETWEEN 1 AND 4096),
+    state TEXT NOT NULL CHECK (state IN ('pending', 'promoted', 'completed', 'abandoned')),
+    pack_name TEXT,
+    created_at INTEGER NOT NULL CHECK (created_at >= 0)
+) STRICT;
+
+CREATE INDEX git_operation_intent_incomplete
+ON git_operation_intent (state, created_at, id)
+WHERE state IN ('pending', 'promoted');
+
+CREATE TABLE git_operation_event (
+    id INTEGER PRIMARY KEY,
+    intent_id TEXT NOT NULL UNIQUE
+        REFERENCES git_operation_intent (id) ON DELETE RESTRICT,
+    payload BLOB NOT NULL
+) STRICT;
+
+INSERT INTO m1a_parent (id, name, created_at)
+VALUES (1, 'fixture', 1);
+
+INSERT INTO m1a_child (id, parent_id, sequence, body, state)
+VALUES (1, 1, 1, 'version three', 'closed');
+
+PRAGMA user_version = 3;

@@ -1,0 +1,547 @@
+#[allow(
+    dead_code,
+    reason = "the SSH push test does not use each authentication API"
+)]
+#[path = "../src/auth.rs"]
+mod auth;
+#[allow(
+    dead_code,
+    reason = "the SSH push test does not use each Git service API"
+)]
+#[path = "../src/git/mod.rs"]
+mod git;
+#[allow(
+    dead_code,
+    reason = "the SSH push test does not inspect the request audit"
+)]
+#[path = "../src/ssh.rs"]
+mod ssh;
+#[allow(dead_code, reason = "the SSH push test does not use each store API")]
+#[path = "../src/store/mod.rs"]
+mod store;
+
+use std::env;
+use std::fs;
+use std::net::{Ipv4Addr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Output};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use auth::SshPublicKey;
+use git::transport::GitRepositories;
+use ssh::RunningSshServer;
+use tempfile::TempDir;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn stock_git_creates_and_updates_branches_for_both_hash_formats_over_ssh() {
+    for format in ["sha1", "sha256"] {
+        let directory = TempDir::new().expect("create an SSH push fixture directory");
+        let repositories_root = directory.path().join("repositories");
+        let bare = repositories_root.join("alice/example.git");
+        let worktree = directory.path().join("worktree");
+        fs::create_dir_all(bare.parent().expect("a bare repository parent"))
+            .expect("create a repository owner directory");
+        run(Command::new("git")
+            .args(["init", "-q", "--bare", "--object-format", format])
+            .arg(&bare));
+        create_worktree(&worktree, format);
+
+        let private_key = directory.path().join("id_ed25519");
+        create_key(&private_key);
+        let key = parse_key(&private_key);
+        let reader_private_key = directory.path().join("reader_ed25519");
+        create_key(&reader_private_key);
+        let reader_key = parse_key(&reader_private_key);
+        let database = directory.path().join("tit.db");
+        store::Store::open(&database).expect("create the intent store");
+        let repositories = GitRepositories::new_with_pushes(&repositories_root, &database)
+            .expect("open the repository root");
+        let server = RunningSshServer::start_with_git_writes(
+            SocketAddr::from((Ipv4Addr::LOCALHOST, 0)),
+            &[key.clone(), reader_key],
+            std::slice::from_ref(&key),
+            repositories,
+        )
+        .await
+        .expect("start the SSH Git server");
+        let url = format!(
+            "ssh://ignored@{}:{}/alice/example",
+            server.address().ip(),
+            server.address().port()
+        );
+
+        push(&worktree, &private_key, &url, &["main"]);
+        assert_eq!(rev_parse(&bare, "refs/heads/main"), head(&worktree));
+
+        fs::write(worktree.join("second.txt"), b"second commit\n").expect("write a second file");
+        run(Command::new("git")
+            .arg("-C")
+            .arg(&worktree)
+            .args(["add", "."]));
+        run(Command::new("git").arg("-C").arg(&worktree).args([
+            "commit",
+            "-q",
+            "-m",
+            "second commit",
+        ]));
+        push(&worktree, &private_key, &url, &["main"]);
+        assert_eq!(rev_parse(&bare, "refs/heads/main"), head(&worktree));
+
+        run(Command::new("git")
+            .arg("-C")
+            .arg(&worktree)
+            .args(["branch", "feature"]));
+        run(Command::new("git")
+            .arg("-C")
+            .arg(&worktree)
+            .args(["tag", "v1"]));
+        push(
+            &worktree,
+            &private_key,
+            &url,
+            &["--atomic", "feature", "v1"],
+        );
+        assert_eq!(
+            rev_parse(&bare, "refs/heads/feature"),
+            rev_parse(&worktree, "refs/heads/feature")
+        );
+        assert_eq!(
+            rev_parse(&bare, "refs/tags/v1"),
+            rev_parse(&worktree, "refs/tags/v1")
+        );
+        push(
+            &worktree,
+            &private_key,
+            &url,
+            &["--delete", "feature", "v1"],
+        );
+        assert_missing_ref(&bare, "refs/heads/feature");
+        assert_missing_ref(&bare, "refs/tags/v1");
+
+        let blob = run(Command::new("git").arg("-C").arg(&worktree).args([
+            "hash-object",
+            "-w",
+            "second.txt",
+        ]));
+        let blob = String::from_utf8(blob.stdout)
+            .expect("read a blob ID")
+            .trim()
+            .to_owned();
+        run(Command::new("git").arg("-C").arg(&worktree).args([
+            "update-ref",
+            "refs/tit/blob",
+            &blob,
+        ]));
+        let invalid_branch = push_result(
+            &worktree,
+            &private_key,
+            &url,
+            &["refs/tit/blob:refs/heads/not-a-commit"],
+        );
+        assert!(
+            !invalid_branch.status.success()
+                && String::from_utf8_lossy(&invalid_branch.stderr)
+                    .contains("branch target is not a commit"),
+            "the server did not reject a branch that targets a blob: {}",
+            String::from_utf8_lossy(&invalid_branch.stderr)
+        );
+        assert_missing_ref(&bare, "refs/heads/not-a-commit");
+
+        let accepted_head = rev_parse(&bare, "refs/heads/main");
+        run(Command::new("git")
+            .arg("-C")
+            .arg(&worktree)
+            .args(["reset", "--hard", "HEAD~1"]));
+        fs::write(worktree.join("rejected.txt"), b"rejected history\n")
+            .expect("write a rejected file");
+        run(Command::new("git")
+            .arg("-C")
+            .arg(&worktree)
+            .args(["add", "."]));
+        run(Command::new("git").arg("-C").arg(&worktree).args([
+            "commit",
+            "-q",
+            "-m",
+            "rejected history",
+        ]));
+        let rejected_object = head(&worktree);
+        let rejected = push_result(&worktree, &private_key, &url, &["--force", "main"]);
+        assert!(
+            !rejected.status.success(),
+            "the server accepted a non-fast-forward push"
+        );
+        assert!(
+            String::from_utf8_lossy(&rejected.stderr).contains("non-fast-forward"),
+            "the server did not return an accurate per-ref status: {}",
+            String::from_utf8_lossy(&rejected.stderr)
+        );
+        assert_eq!(rev_parse(&bare, "refs/heads/main"), accepted_head);
+        assert_missing_object(&bare, &rejected_object);
+        let atomic_rejection = push_result(
+            &worktree,
+            &private_key,
+            &url,
+            &[
+                "--atomic",
+                "--force",
+                "main",
+                "HEAD:refs/heads/must-not-exist",
+            ],
+        );
+        assert!(
+            !atomic_rejection.status.success(),
+            "the server partially accepted an invalid atomic push"
+        );
+        assert_missing_ref(&bare, "refs/heads/must-not-exist");
+        assert_eq!(rev_parse(&bare, "refs/heads/main"), accepted_head);
+        assert!(
+            store::Store::open(&database)
+                .expect("open the intent store after rejection")
+                .incomplete_git_intents()
+                .expect("list incomplete rejected pushes")
+                .is_empty(),
+            "a rejected push left an incomplete intent"
+        );
+        assert_eq!(
+            fs::read_dir(bare.join("objects/tit-quarantine"))
+                .map(|entries| entries.count())
+                .unwrap_or(0),
+            0,
+            "a rejected push left a quarantine directory"
+        );
+
+        let unauthorized = push_result(
+            &worktree,
+            &reader_private_key,
+            &url,
+            &["HEAD:refs/heads/unauthorized"],
+        );
+        assert!(
+            !unauthorized.status.success(),
+            "the server accepted a push from a read-only key"
+        );
+        assert_missing_ref(&bare, "refs/heads/unauthorized");
+
+        server.shutdown().await.expect("stop the SSH Git server");
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn git_push_crash_child() {
+    let Some(mode) = env::var_os("TIT_M1D_CHILD_MODE") else {
+        return;
+    };
+    let repositories_root = required_path("TIT_M1D_REPOSITORIES");
+    let database = required_path("TIT_M1D_DATABASE");
+    let worktree = required_path("TIT_M1D_WORKTREE");
+    let private_key = required_path("TIT_M1D_PRIVATE_KEY");
+    let key = parse_key(&private_key);
+    let repositories = GitRepositories::new_with_pushes(&repositories_root, &database)
+        .expect("open the crash-test repository root");
+    let server = RunningSshServer::start_with_git_writes(
+        SocketAddr::from((Ipv4Addr::LOCALHOST, 0)),
+        std::slice::from_ref(&key),
+        std::slice::from_ref(&key),
+        repositories,
+    )
+    .await
+    .expect("start the crash-test SSH server");
+    let url = format!(
+        "ssh://ignored@{}:{}/alice/example",
+        server.address().ip(),
+        server.address().port()
+    );
+    let output = push_result(&worktree, &private_key, &url, &["main"]);
+    panic!(
+        "the {mode:?} crash hook did not stop receive-pack: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+}
+
+#[test]
+fn reconciles_process_termination_at_each_cross_store_boundary() {
+    for boundary in ["intent", "objects", "refs", "completed"] {
+        let directory = TempDir::new().expect("create a recovery fixture directory");
+        let repositories_root = directory.path().join("repositories");
+        let bare = repositories_root.join("alice/example.git");
+        let worktree = directory.path().join("worktree");
+        fs::create_dir_all(bare.parent().expect("a bare repository parent"))
+            .expect("create a repository owner directory");
+        run(Command::new("git")
+            .args(["init", "-q", "--bare", "--object-format", "sha1"])
+            .arg(&bare));
+        create_worktree(&worktree, "sha1");
+        let proposed = head(&worktree);
+        let private_key = directory.path().join("id_ed25519");
+        create_key(&private_key);
+        let database = directory.path().join("tit.db");
+        store::Store::open(&database).expect("create the intent store");
+        let ready = directory.path().join("ready");
+
+        let child = spawn_push_crash_child(
+            boundary,
+            &repositories_root,
+            &database,
+            &worktree,
+            &private_key,
+            &ready,
+        );
+        kill_after_ready(child, &ready);
+        git::receive_pack::recover_incomplete_pushes(&database)
+            .expect("recover the interrupted push");
+
+        let store = store::Store::open(&database).expect("open the recovered intent store");
+        assert!(
+            store
+                .incomplete_git_intents()
+                .expect("list incomplete intents")
+                .is_empty(),
+            "{boundary} left an incomplete intent"
+        );
+        let events = store
+            .connection()
+            .query_row("SELECT count(*) FROM git_operation_event", [], |row| {
+                row.get::<_, i64>(0)
+            })
+            .expect("count Git events");
+        if matches!(boundary, "refs" | "completed") {
+            assert_eq!(rev_parse(&bare, "refs/heads/main"), proposed);
+            assert_eq!(events, 1);
+        } else {
+            assert_missing_ref(&bare, "refs/heads/main");
+            assert_missing_object(&bare, &proposed);
+            assert_eq!(events, 0);
+        }
+        let quarantine = bare.join("objects/tit-quarantine");
+        let entries = fs::read_dir(&quarantine)
+            .map(|entries| entries.count())
+            .unwrap_or(0);
+        assert_eq!(entries, 0, "{boundary} left a quarantine directory");
+    }
+}
+
+#[test]
+fn fuzzed_packet_lines_and_pack_inputs_do_not_panic() {
+    let directory = TempDir::new().expect("create a fuzz fixture directory");
+    let repositories_root = directory.path().join("repositories");
+    let bare = repositories_root.join("alice/example.git");
+    fs::create_dir_all(bare.parent().expect("a bare repository parent"))
+        .expect("create a repository owner directory");
+    run(Command::new("git")
+        .args(["init", "-q", "--bare", "--object-format", "sha1"])
+        .arg(&bare));
+    let database = directory.path().join("tit.db");
+    store::Store::open(&database).expect("create the intent store");
+    let mut command = Vec::new();
+    git::packetline::encode_data(
+        b"0000000000000000000000000000000000000000 1111111111111111111111111111111111111111 refs/heads/main\0 report-status\n",
+        &mut command,
+    )
+    .expect("encode a receive-pack command");
+    git::packetline::encode_flush(&mut command);
+
+    let mut state = 0x9e37_79b9_7f4a_7c15_u64;
+    for length in 0..256 {
+        let mut bytes = Vec::with_capacity(length);
+        for _ in 0..length {
+            state ^= state << 13;
+            state ^= state >> 7;
+            state ^= state << 17;
+            bytes.push(state as u8);
+        }
+        std::panic::catch_unwind(|| {
+            let _ = git::packetline::decode(&bytes);
+            let _ = git::packetline::first_flush_end(&bytes);
+        })
+        .expect("packet-line parsing must not panic");
+
+        let mut receive =
+            git::receive_pack::ReceivePack::open(&bare, &database, "SHA256:fuzz-key".to_owned())
+                .expect("open receive-pack for fuzz input");
+        fs::write(receive.incoming_pack(), &bytes).expect("write a fuzz pack");
+        std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _ = receive.finish(&command);
+        }))
+        .expect("pack parsing must not panic");
+    }
+    git::receive_pack::recover_incomplete_pushes(&database)
+        .expect("recover rejected fuzz operations");
+}
+
+fn spawn_push_crash_child(
+    boundary: &str,
+    repositories: &Path,
+    database: &Path,
+    worktree: &Path,
+    private_key: &Path,
+    ready: &Path,
+) -> Child {
+    Command::new(env::current_exe().expect("find the integration test executable"))
+        .args([
+            "--exact",
+            "git_push_crash_child",
+            "--nocapture",
+            "--test-threads=1",
+        ])
+        .env("TIT_M1D_CHILD_MODE", boundary)
+        .env("TIT_M1D_CRASH_AFTER", boundary)
+        .env("TIT_M1D_REPOSITORIES", repositories)
+        .env("TIT_M1D_DATABASE", database)
+        .env("TIT_M1D_WORKTREE", worktree)
+        .env("TIT_M1D_PRIVATE_KEY", private_key)
+        .env("TIT_M1D_READY", ready)
+        .spawn()
+        .expect("start the push crash-test child")
+}
+
+fn kill_after_ready(mut child: Child, ready: &Path) {
+    let deadline = Instant::now() + Duration::from_secs(15);
+    while !ready.exists() {
+        if let Some(status) = child.try_wait().expect("inspect the push crash-test child") {
+            panic!("push crash-test child stopped before it was ready: {status}");
+        }
+        assert!(
+            Instant::now() < deadline,
+            "push crash-test child was not ready"
+        );
+        thread::sleep(Duration::from_millis(10));
+    }
+    child.kill().expect("kill the push crash-test child");
+    child.wait().expect("wait for the push crash-test child");
+}
+
+fn required_path(name: &str) -> PathBuf {
+    env::var_os(name)
+        .map(PathBuf::from)
+        .unwrap_or_else(|| panic!("read {name}"))
+}
+
+fn create_worktree(path: &Path, format: &str) {
+    run(Command::new("git")
+        .args(["init", "-q", "--object-format", format, "-b", "main"])
+        .arg(path));
+    for (name, value) in [
+        ("user.name", "Tit Test"),
+        ("user.email", "tit@example.invalid"),
+        ("commit.gpgsign", "false"),
+        ("tag.gpgsign", "false"),
+    ] {
+        run(Command::new("git")
+            .arg("-C")
+            .arg(path)
+            .args(["config", name, value]));
+    }
+    fs::write(path.join("first.txt"), b"first commit\n").expect("write the first file");
+    run(Command::new("git").arg("-C").arg(path).args(["add", "."]));
+    run(Command::new("git")
+        .arg("-C")
+        .arg(path)
+        .args(["commit", "-q", "-m", "first commit"]));
+}
+
+fn create_key(path: &Path) {
+    run(Command::new("ssh-keygen")
+        .args(["-q", "-t", "ed25519", "-N", "", "-f"])
+        .arg(path));
+}
+
+fn parse_key(private_key: &Path) -> SshPublicKey {
+    let mut public_key = private_key.as_os_str().to_owned();
+    public_key.push(".pub");
+    let encoded = fs::read_to_string(PathBuf::from(public_key)).expect("read the public key");
+    SshPublicKey::parse(&encoded).expect("parse the public key")
+}
+
+fn push(worktree: &Path, private_key: &Path, url: &str, refs: &[&str]) -> Output {
+    let output = push_result(worktree, private_key, url, refs);
+    assert!(
+        output.status.success(),
+        "stock Git push failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    output
+}
+
+fn push_result(worktree: &Path, private_key: &Path, url: &str, refs: &[&str]) -> Output {
+    Command::new("git")
+        .arg("-C")
+        .arg(worktree)
+        .arg("push")
+        .arg(url)
+        .args(refs)
+        .env("GIT_SSH_COMMAND", ssh_command(private_key))
+        .env("GIT_SSH_VARIANT", "ssh")
+        .env("GIT_TRACE_PACKET", "1")
+        .output()
+        .expect("run stock Git push")
+}
+
+fn ssh_command(private_key: &Path) -> String {
+    format!(
+        "ssh -F /dev/null -i {} -o BatchMode=yes -o IdentitiesOnly=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR -o PasswordAuthentication=no -o KbdInteractiveAuthentication=no",
+        private_key.display()
+    )
+}
+
+fn head(path: &Path) -> String {
+    rev_parse(path, "HEAD")
+}
+
+fn rev_parse(path: &Path, name: &str) -> String {
+    let output = Command::new("git")
+        .arg("--git-dir")
+        .arg(path)
+        .args(["rev-parse", name])
+        .output()
+        .expect("run stock Git rev-parse");
+    let output = if output.status.success() {
+        output
+    } else {
+        Command::new("git")
+            .arg("-C")
+            .arg(path)
+            .args(["rev-parse", name])
+            .output()
+            .expect("run worktree Git rev-parse")
+    };
+    assert!(
+        output.status.success(),
+        "rev-parse {name} failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    String::from_utf8(output.stdout)
+        .expect("read an object ID")
+        .trim()
+        .to_owned()
+}
+
+fn assert_missing_ref(repository: &Path, name: &str) {
+    let output = Command::new("git")
+        .arg("--git-dir")
+        .arg(repository)
+        .args(["show-ref", "--verify", "--quiet", name])
+        .output()
+        .expect("inspect a missing ref");
+    assert!(!output.status.success(), "{name} still exists");
+}
+
+fn assert_missing_object(repository: &Path, id: &str) {
+    let output = Command::new("git")
+        .arg("--git-dir")
+        .arg(repository)
+        .args(["cat-file", "-e", id])
+        .output()
+        .expect("inspect a rejected object");
+    assert!(!output.status.success(), "rejected object {id} is visible");
+}
+
+fn run(command: &mut Command) -> Output {
+    let output = command.output().expect("run stock command");
+    assert!(
+        output.status.success(),
+        "stock command failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    output
+}

@@ -16,6 +16,9 @@
 )]
 #[path = "../src/ssh.rs"]
 mod ssh;
+#[allow(dead_code, reason = "the SSH Git test does not use the intent store")]
+#[path = "../src/store/mod.rs"]
+mod store;
 
 use std::fs;
 use std::net::{Ipv4Addr, SocketAddr};

@@ -9,7 +9,7 @@
 use std::{env, ffi::OsString, fs};
 
 use rusqlite::{Connection, ErrorCode, TransactionBehavior, params};
-use store::{Store, StoreError};
+use store::{GitOperationIntent, Store, StoreError};
 use tempfile::TempDir;
 
 const V1_FIXTURE: &str = include_str!("fixtures/sqlite/v1.sql");
@@ -131,7 +131,7 @@
     let directory = TempDir::new().expect("create a temporary directory");
     let store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
 
-    assert_eq!(store.schema_version().expect("read the schema version"), 2);
+    assert_eq!(store.schema_version().expect("read the schema version"), 3);
     assert_eq!(
         store
             .connection()
@@ -161,6 +161,77 @@
         5_000
     );
     store.integrity_check().expect("check database integrity");
+}
+
+#[test]
+fn persists_and_completes_git_operation_intents_with_their_events() {
+    let directory = TempDir::new().expect("create a temporary directory");
+    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
+    let intent = GitOperationIntent {
+        id: "00112233445566778899aabbccddeeff",
+        repository_path: "/srv/tit/repositories/alice/example.git",
+        actor: "SHA256:test-key",
+        initial_refs: b"0000000000000000000000000000000000000000 refs/heads/main\n",
+        proposed_refs: b"1111111111111111111111111111111111111111 refs/heads/main\n",
+        event_payload: b"push event",
+        quarantine_path: "/srv/tit/repositories/alice/example.git/objects/tit-quarantine/test",
+        created_at: 1,
+    };
+    store.begin_git_intent(&intent).expect("begin an intent");
+    let pending = store
+        .incomplete_git_intents()
+        .expect("list pending intents");
+    assert_eq!(pending.len(), 1);
+    assert_eq!(pending[0].id, intent.id);
+    assert_eq!(pending[0].repository_path, intent.repository_path);
+    assert_eq!(pending[0].initial_refs, intent.initial_refs);
+    assert_eq!(pending[0].proposed_refs, intent.proposed_refs);
+    assert_eq!(pending[0].quarantine_path, intent.quarantine_path);
+    assert_eq!(pending[0].state, "pending");
+    assert_eq!(pending[0].pack_name, None);
+
+    store
+        .mark_git_objects_promoted(intent.id, Some("pack-test.pack"))
+        .expect("mark objects as promoted");
+    let promoted = store
+        .incomplete_git_intents()
+        .expect("list promoted intents");
+    assert_eq!(promoted[0].state, "promoted");
+    assert_eq!(promoted[0].pack_name.as_deref(), Some("pack-test.pack"));
+    store
+        .complete_git_intent(intent.id)
+        .expect("complete the intent");
+    assert!(
+        store
+            .incomplete_git_intents()
+            .expect("list incomplete intents")
+            .is_empty()
+    );
+    assert_eq!(
+        store
+            .connection()
+            .query_row("SELECT count(*) FROM git_operation_event", [], |row| row
+                .get::<_, i64>(0))
+            .expect("count Git operation events"),
+        1
+    );
+
+    let abandoned = GitOperationIntent {
+        id: "ffeeddccbbaa99887766554433221100",
+        ..intent
+    };
+    store
+        .begin_git_intent(&abandoned)
+        .expect("begin an abandoned intent");
+    store
+        .abandon_git_intent(abandoned.id)
+        .expect("abandon the intent");
+    assert!(
+        store
+            .incomplete_git_intents()
+            .expect("list incomplete intents")
+            .is_empty()
+    );
 }
 
 #[test]
@@ -496,7 +567,7 @@
         create_fixture(&path, fixture);
 
         let store = Store::open(&path).expect("migrate the fixture");
-        assert_eq!(store.schema_version().expect("read the schema version"), 2);
+        assert_eq!(store.schema_version().expect("read the schema version"), 3);
         store.integrity_check().expect("check migrated integrity");
         let state: String = store
             .connection()
@@ -512,19 +583,20 @@
         assert_eq!(state, expected);
         store::doctor(directory.path()).expect("check the migrated fixture with doctor");
 
-        let backup_path = migration_backup(&path, 1);
-        assert_eq!(backup_path.exists(), initial_version == 1);
-        if initial_version == 1 {
-            let backup = Store::open_unmigrated(&backup_path).expect("open the migration backup");
-            assert_eq!(backup.schema_version().expect("read the backup version"), 1);
-            backup.integrity_check().expect("check backup integrity");
-        }
+        let backup_path = migration_backup(&path, initial_version);
+        assert!(backup_path.exists());
+        let backup = Store::open_unmigrated(&backup_path).expect("open the migration backup");
+        assert_eq!(
+            backup.schema_version().expect("read the backup version"),
+            initial_version
+        );
+        backup.integrity_check().expect("check backup integrity");
     }
 }
 
 #[test]
 fn recovers_complete_schema_versions_after_a_process_kill_during_migration() {
-    for (mode, expected_version) in [("migration-uncommitted", 1), ("migration-committed", 2)] {
+    for (mode, expected_version) in [("migration-uncommitted", 1), ("migration-committed", 3)] {
         let directory = TempDir::new().expect("create a temporary directory");
         let path = database(&directory, "fixture.sqlite");
         create_fixture(&path, V1_FIXTURE);
@@ -549,7 +621,7 @@
                 |row| row.get(0),
             )
             .expect("inspect the recovered schema");
-        assert_eq!(has_state, i64::from(expected_version == 2));
+        assert_eq!(has_state, i64::from(expected_version >= 2));
     }
 }
 

@@ -1,3 +1,7 @@
+#[allow(
+    dead_code,
+    reason = "the M1A workload does not use Git operation intents"
+)]
 #[path = "../src/store/mod.rs"]
 mod store;
 

@@ -13,6 +13,12 @@
 )]
 #[path = "../src/ssh.rs"]
 mod ssh;
+#[allow(
+    dead_code,
+    reason = "the SSH identity test does not use the intent store"
+)]
+#[path = "../src/store/mod.rs"]
+mod store;
 
 use std::fs;
 use std::net::{Ipv4Addr, SocketAddr};