Race condition v SafeStream?

	define('WORKERS', 2);
	define('MESSAGES', 10);
	define('FILE_PATH', 'test_SafeStream_' . rand() . '.txt');

	$children = [];
	for($i = 0; $i < WORKERS; ++$i) {
		$pid = pcntl_fork();
		if( $pid == -1 ) {
			echo "Fork failed, bailing out...\n";
			break;
		} elseif( $pid == 0 ) {
			// child

			// "seed" lcg_value() generator so SafeStream doesn't generate
			// the same .tmp filename (and then fails to open it)
			for($j = 0; $j < (WORKERS - $i) * MESSAGES; ++$j)
				lcg_value();

			for($j = 0; $j < MESSAGES; ++$j) {
				// open file for append, write message and close

				if( !$f = fopen('nette.safe://' . FILE_PATH, 'a') )
					die("Worker $i failed to open " . FILE_PATH . " for message $j\n");

				$msg =  "Worker $i, message $j\n";
				$r = fwrite($f, $msg);
				if( $r === false || $r != strlen($msg) )
					die("Worker $i failed to append message $j\n");

				if( !fclose($f) )
					die("Worker $i failed to close file properly\n");
			}

			// child proces ends here
			exit(0);
		} else {
			// main process

			$children[] = $pid;
		}
	}

	while( !empty($children) ) {
		$child_pid = pcntl_wait($status);

		$children = array_filter($children, fn($pid) => $pid != $child_pid);
	}

	echo "Expected " . (WORKERS * MESSAGES) . " messages, got " . substr_count(file_get_contents(FILE_PATH), "\n") . "\n";

	unlink(FILE_PATH);

class EvenSaferStream
{
	/** Name of stream protocol - nette.safe:// */
	public const PROTOCOL = 'nette.safe';

	private $path;

	private $work_path;
	private $work_fd;

	private $lock_path;
	private $lock_fd;

	private const OK = 0, Error = -1, Blocked = -2;


	public static function register(): bool
	{
		return stream_wrapper_register(self::PROTOCOL, self::class);
	}


	public function stream_open(string $path, string $mode, int $options): bool
	{
		$path = substr($path, strpos($path, ':') + 3);  // trim protocol nette.safe://

		if( $mode != 'a' ) // support only append
			return false;

		if( $this->lock($path) != self::OK )
			return false;

		$ok = false;
		try {
			$work_path = $path . '.tmp';

			if( file_exists($path) ) {
				if( !$fd = fopen($path, 'r+') )
					return false;

				$copy = stream_get_contents($fd);
				if( $copy === false )
					return false;

				if( file_put_contents($work_path, $copy) !== strlen($copy) )
					return false;

				fclose($fd);
			}

			if( !$work_fd = fopen($work_path, 'a') ) {
				@ unlink($work_path);
				return false;
			}

			$this->path = $path;
			$this->work_path = $work_path;
			$this->work_fd = $work_fd;

			$ok = true;

			return true;
		} finally {
			if( !$ok ) {
				$this->unlock();
			}
		}
	}

	private const Attempts = 10, TokenLength = 10;

	private function lock($path)
	{
		$lock_path = $path . '.lock';

		for($i=0; $i < self::Attempts; ++$i) {
			if( !$lock_fd = fopen($lock_path, 'c') ) {
				return self::Error;
			}

			if( !flock($lock_fd, LOCK_EX) ) {
				fclose($lock_fd);
				return self::Error;
			}

			if( !$stat = fstat($lock_fd) ) {
				fclose($lock_fd);
				return self::Error;
			}

			if( $stat['size'] > 0 ) {
				// op on this lock fd was successfully committed, but we need to
				// check if the lock file was also removed.
				if( file_exists($lock_path) ) {
					// Lock file is still there, so now we need to figure out if $lock_path points
					// to our $lock_fd or it's new (which means there is another process with lock() on this path).
					// We will do it by writing random challenge string into $lock_fd and see if
					// it appears in $lock_path file.
					$challenge = random_bytes(self::TokenLength);
					fwrite($lock_fd, $challenge);
					fflush($lock_fd);

					if( $challenge === @ file_get_contents($lock_path) )
						unlink($lock_path);
				}

				fclose($lock_fd);
				continue;
			}

			$this->lock_path = $lock_path;
			$this->lock_fd = $lock_fd;

			return self::OK;
		}

		return self::Blocked;
	}

	private function unlock()
	{
		fwrite($this->lock_fd, 'c'); // signal we released the file correctly
		fflush($this->lock_fd);
		unlink($this->lock_path);
		fclose($this->lock_fd); // release lock

		$this->lock_path = null;
		$this->lock_fd = null;
	}


	public function stream_close(): void
	{
		fclose($this->work_fd);
		rename($this->work_path, $this->path);

		$this->work_path = null;
		$this->work_fd = null;

		$this->unlock();
	}

	public function stream_read(int $length)
	{
		return fread($this->work_fd, $length);
	}

	public function stream_write(string $data)
	{
		return fwrite($this->work_fd, $data, strlen($data));
	}

	public function stream_truncate(int $size): bool
	{
		return ftruncate($this->work_fd, $size);
	}

	public function stream_tell(): int
	{
		return ftell($this->work_fd);
	}

	public function stream_eof(): bool
	{
		return feof($this->work_fd);
	}

	public function stream_seek(int $offset, int $whence = SEEK_SET): bool
	{
		return fseek($this->work_fd, $offset, $whence) === 0;
	}

	public function stream_stat()
	{
		return fstat($this->work_fd);
	}
}

<?php

class EvenSaferStream
{
	/** Name of stream protocol - nette.safe:// */
	public const PROTOCOL = 'nette.safe';

	// True if opened in r mode.
	private bool $readOnly;

	// Path to the file.
	private ?string $path;

	// Path & handle to working file that will be moved to $path at the end,
	// except in case of read only mode, where $workHandle points to opened
	// $path and $workPath is null.
	private ?string $workPath;
	private $workHandle;

	// Path & handle to lock, if opened in any other mode than read only.
	private ?string $lockPath;
	private $lockHandle;


	private static function getTemporaryPath($path)
	{
		$dir = dirname($path);
		return ($dir !== '' ? $dir . '/' : '') . '~~' . basename($path);
	}

	// Locking is done on separate $path.lock file, which preferably should be removed after
	// work is over. But it also should be there as long as possible to not disturb the FIFO
	// front of processes waiting on it for their turn. For that, before attempt to acquire
	// lock is made, one byte is appended to the opened lock file. This is done without
	// synchronization and some data may be lost, so after acquiring lock the process needs
	// to check $lock_path still points to the opened lock handle and it wasn't unlinked by
	// previous lock holder who thought there is no one else waiting.

	private const OK = 0, Error = -1, Blocked = -2;
	private const Attempts = 3;

	private function lock($path)
	{
		$lock_path = self::getTemporaryPath($path) . '.lock';

		for($i=0; $i < self::Attempts; ++$i) {
			if( !$lock_handle = fopen($lock_path, 'a') )
				return self::Error;

			// Append hint to lock file signaling there is someone waiting on it.
			fwrite($lock_handle, 'o');
			fflush($lock_handle);

			if( !flock($lock_handle, LOCK_EX) ) {
				fclose($lock_handle);
				return self::Error;
			}

			$lock_stat = fstat($lock_handle);

			// Decrement string of waiting hints.
			ftruncate($lock_handle, $lock_stat['size'] > 0 ? $lock_stat['size'] - 1 : 0);

			// Now check the lock we acquired is the same as under $lock_path. If inodes
			// are different (or $lock_path doesn't exists), it means someone else deleted
			// the lock and we have to throw it away and start again from current $lock_path.

			clearstatcache(true, $lock_path);

			$lock_path_ino = @ fileinode($lock_path);
			if( $lock_path_ino === false || $lock_path_ino != $lock_stat['ino'] ) {
				fclose($lock_handle);
				continue;
			}

			$this->lockPath = $lock_path;
			$this->lockHandle = $lock_handle;

			return self::OK;
		}

		return self::Blocked;
	}

	private function unlock()
	{
		$others_wait_on_lock = fstat($this->lockHandle)['size'] > 0;

		// This is rough approximation, it's possible there are others waiting on
		// the lock even if this says there are none. However when they wake up
		// they should detect that in lock() and run another attempt.
		if( !$others_wait_on_lock )
			unlink($this->lockPath);

		fclose($this->lockHandle); // release the lock

		$this->lockPath = null;
		$this->lockHandle = null;
	}


	public static function register(): bool
	{
		return stream_wrapper_register(self::PROTOCOL, self::class);
	}

	public function stream_open(string $path, string $mode, int $options): bool
	{
		$path = substr($path, strlen(self::PROTOCOL) + 3);  // trim protocol nette.safe://

		$tb_flag = ltrim($mode, 'acrxw+');
		$mode = rtrim($mode, 'tb');
		$use_include_path = (bool) (STREAM_USE_PATH & $options);

		if( $mode == 'r' ) {
			// for read only mode no need to lock anything, just use what is there
			// under $path
			if( !$fd = fopen($path, 'r' . $tb_flag, $use_include_path) )
				return false;

			$this->readOnly = true;
			$this->workHandle = $fd;

			return true;
		}

		$this->readOnly = false;

		if( $this->lock($path) != self::OK )
			return false;

		$ok = false;
		try {
			$work_path = self::getTemporaryPath($path) . '.tmp';

			if( $mode[0] != 'w' && $mode[0] != 'x' ) {
				// w & x mode truncate file, others don't => need to copy content
				// to working file
				clearstatcache(true, $path);

				// FIXME: this doesn't use $use_include_path
				if( file_exists($path) ) {
					if( !copy($path, $work_path) )
						return false;
				} elseif( $mode[0] == 'r' ) {
					return false;
				}
			} elseif( $mode[0] == 'x' ) {
				// make sure we are first to create $path
				if( !fopen($path, $mode . $tb_flag, $use_include_path) )
					return false;
			}

			if( !$work_handle = fopen($work_path, $mode . $tb_flag, $use_include_path) ) {
				@ unlink($work_path);
				return false;
			}

			$this->path = $path;
			$this->workPath = $work_path;
			$this->workHandle = $work_handle;

			$ok = true;

			return true;
		} finally {
			if( !$ok ) {
				$this->unlock();
			}
		}
	}

	public function stream_close(): void
	{
		fclose($this->workHandle);

		if( $this->readOnly )
			return;

		rename($this->workPath, $this->path);

		$this->workPath = null;
		$this->workHandle = null;

		$this->unlock();
	}

	public function stream_read(int $length)
	{
		return fread($this->workHandle, $length);
	}

	public function stream_write(string $data)
	{
		return fwrite($this->workHandle, $data, strlen($data));
	}

	public function stream_truncate(int $size): bool
	{
		return ftruncate($this->workHandle, $size);
	}

	public function stream_tell(): int
	{
		return ftell($this->workHandle);
	}

	public function stream_eof(): bool
	{
		return feof($this->workHandle);
	}

	public function stream_seek(int $offset, int $whence = SEEK_SET): bool
	{
		return fseek($this->workHandle, $offset, $whence) === 0;
	}

	public function stream_stat()
	{
		return fstat($this->workHandle);
	}
}

?>