Several web processes can share a data directory (closes #274) ec84907043c2

- **Several web processes can share a data directory (#274):** the transparency log took appends from one process only, so only one process per data directory could have the web role. Each append now holds a lock in the log's directory and starts from the log as the files hold it, readers see every process's entries, and every process serves the newest checkpoint, whichever signed it.

    /// `append.lock`: held, exclusively, around each append, so several

    /// processes sharing the data directory append one at a time (#274).

    lock: File,

    /// process without it only reads, catching up with the writers.

    /// are, so it never cuts short an append a writer is making. Several

    /// web processes may share the data directory and write (#274).

        let lock = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(dir.join("append.lock"))?;

        let mut files = Files { entries, offsets, hashes, size: 0, entries_len: 0 };

        if writer {

            lock.lock()?;

            let done = Self::recover(&mut files, true);

            lock.unlock()?;

            done?;

        } else {

            Self::recover(&mut files, false)?;

        }

        let size = files.size;

        Ok(Log {

            dir: dir.to_path_buf(),

            origin: origin.to_string(),

            files: Mutex::new(files),

            lock,

            checkpoint: Mutex::new(Self::read_checkpoint(dir, size)),

            witnesses: Default::default(),

            witness_for: Default::default(),

            writer,

        })

    }

    /// Read the log's extent from its files, as another process may have

    /// appended since (#274). The whole entries count: those whose offset

    /// and bytes were written, and whose stored hashes all were. A writer,

    /// holding the append lock, also finishes what an append cut short left

    /// (it truncates partial bytes and computes missing hashes again); a

    /// reader changes nothing and takes the whole entries only.

    fn recover(files: &mut Files, writer: bool) -> anyhow::Result<()> {

        let entries_on_disk = files.entries.metadata()?.len();

        let mut size = files.offsets.metadata()?.len() / 8;

            let off = read_u64_at(&files.offsets, (size - 1) * 8)?;

            if off + 2 <= entries_on_disk && files.entries.read_exact_at(&mut len, off).is_ok() {

        let have = files.hashes.metadata()?.len() / 32;

            files.size = whole;

            files.entries_len = entries_len;

            return Ok(());

        files.offsets.set_len(size * 8)?;

        files.entries.set_len(entries_len)?;

        files.hashes.set_len(tlog::stored_hash_count(whole) * 32)?;

        files.entries_len = entries_len;

            let bytes = Self::entry_bytes(files, n)?;

        Ok(())

    }

    /// Catch up with what other processes appended (#274).

    fn catch_up(&self, f: &mut Files) {

        if let Err(e) = Self::recover(f, false) {

            tracing::warn!("tlog: reading the log's extent: {e:#}");

        }

        let mut f = self.files.lock().unwrap();

        if n >= f.size {

            self.catch_up(&mut f);

        }

        let mut f = self.files.lock().unwrap();

        self.catch_up(&mut f);

        f.size

        // One append at a time across the processes sharing the directory

        // (#274), each from the log as the last left it.

        self.lock.lock()?;

        let appended = Self::recover(&mut f, true).and_then(|()| Self::append_locked(&mut f, entries));

        self.lock.unlock()?;

        appended

    }

    fn append_locked(f: &mut Files, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {

    /// The latest signed checkpoint note: the newer of this process's and

    /// the one kept in the directory, which the process leading the

    /// checkpointer signs (#274).

        let mut held = self.checkpoint.lock().unwrap();

        if let Some(kept) = Self::read_checkpoint(&self.dir, u64::MAX) {

            if held.as_ref().is_none_or(|(s, _)| kept.0 > *s) {

                *held = Some(kept);

            }

        }

        held.as_ref().map(|(_, n)| n.clone())

        // From the newest kept, whichever process signed it.

        let _ = self.checkpoint();

    }

    /// oxygenhub#274: writers sharing the directory (each its own handles,

    /// as separate processes have) append one at a time, each from the log

    /// as the others left it; readers and checkpoints follow.

    #[test]

    fn writers_sharing_the_directory_append_one_at_a_time() {

        let dir = tempfile::tempdir().unwrap();

        let a = std::sync::Arc::new(Log::open(dir.path(), "hub.example/log").unwrap());

        let b = std::sync::Arc::new(Log::open(dir.path(), "hub.example/log").unwrap());

        let reader = Log::open_reader(dir.path(), "hub.example/log").unwrap();

        let spawn = |log: std::sync::Arc<Log>, from: u64| {

            std::thread::spawn(move || {

                let mut got = Vec::new();

                for i in 0..150 {

                    got.extend(log.append(&[entry(from + i)]).unwrap());

                }

                got

            })

        };

        let (ta, tb) = (spawn(a.clone(), 0), spawn(b.clone(), 1000));

        let mut numbers: Vec<u64> = ta.join().unwrap().into_iter().chain(tb.join().unwrap()).collect();

        numbers.sort();

        assert_eq!(numbers, (0..300).collect::<Vec<_>>(), "every entry its own place");

        assert_eq!((a.size(), b.size(), reader.size()), (300, 300, 300));

        let root = a.tree_hash(300).unwrap();

        assert_eq!(b.tree_hash(300).unwrap(), root);

        let fresh = Log::open(dir.path(), "hub.example/log").unwrap();

        assert_eq!(fresh.tree_hash(300).unwrap(), root, "the files are one log");

        assert_eq!(b.entry(299).unwrap(), fresh.entry(299).unwrap());

        assert_eq!(reader.entry(299).unwrap(), fresh.entry(299).unwrap());

        // A checkpoint one signs, the other serves.

        let sk = oxy_core::identity::SecretKey::generate();

        let hub = sk.identity("hub", "[email protected]", 1);

        assert!(a.refresh_checkpoint(&hub, &sk).unwrap());

        assert_eq!(b.checkpoint(), a.checkpoint());

        assert!(!b.refresh_checkpoint(&hub, &sk).unwrap(), "already covered by the other's");

transparency log takes appends from every web process sharing its directory (#274).

- **Several web processes may share a data directory (#274).** Each

  append to the transparency log holds `append.lock` in the log's

  directory exclusively, and starts from the log as the files hold it (it

  finishes what a crashed append left, then appends); readers catch up

  with every writer's entries, and the checkpoint is read from its file,

  signed by whichever process leads `tlog.checkpointer`. Across machines

  that needs the data directory on one shared filesystem with working

  `flock`; a log role the web nodes call is the answer for that, later.

  Any number may be workers.