diff --git a/pysyncobj/atomic_replace.py b/pysyncobj/atomic_replace.py index c8a4ae2..a4f3f4a 100644 --- a/pysyncobj/atomic_replace.py +++ b/pysyncobj/atomic_replace.py @@ -30,4 +30,17 @@ def atomicReplace(oldPath, newPath): CloseHandle(ta) return bool(res) else: - atomicReplace = os.rename + def atomicReplace(oldPath, newPath): + # Journal compaction may follow immediately after snapshot publication. + # Persist the contents before rename and the directory before success. + with open(oldPath, 'rb') as f: + os.fsync(f.fileno()) + + os.rename(oldPath, newPath) + + directory = os.path.dirname(os.path.abspath(newPath)) + fd = os.open(directory, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) diff --git a/test_syncobj.py b/test_syncobj.py index 8be7ac7..803d69c 100755 --- a/test_syncobj.py +++ b/test_syncobj.py @@ -1,5 +1,6 @@ from __future__ import print_function import os +import stat import time import pytest import random @@ -20,6 +21,8 @@ from pysyncobj.syncobj_admin import executeAdminCommand from pysyncobj.batteries import ReplCounter, ReplList, ReplDict, ReplSet, ReplLockManager, ReplQueue, ReplPriorityQueue from pysyncobj.node import TCPNode +from pysyncobj.serializer import Serializer +from pysyncobj.config import SERIALIZER_STATE from collections import defaultdict logging.basicConfig(format=u'[%(asctime)s %(filename)s:%(lineno)d %(levelname)s] %(message)s', level=logging.DEBUG) @@ -591,6 +594,77 @@ def test_checkDumpToFile(): checkDumpToFile(False) +def storeSnapshot(serializer, data, incoming=False): + if incoming: + source = Serializer(None, 65536, False, None, None, None) + source.serialize(data, 1) + assert source.checkSerializing()[0] == SERIALIZER_STATE.SUCCESS + assert not serializer.setTransmissionData(source.getTransmissionData(1)) + return serializer.setTransmissionData(source.getTransmissionData(1)) + + serializer.serialize(data, 1) + deadline = time.time() + 10 + while time.time() < deadline: + state, _ = serializer.checkSerializing() + if state != SERIALIZER_STATE.SERIALIZING: + assert state in (SERIALIZER_STATE.SUCCESS, SERIALIZER_STATE.FAILED) + return state == SERIALIZER_STATE.SUCCESS + time.sleep(0.01) + pytest.fail('Snapshot serializer did not finish') + + +@pytest.mark.skipif(os.name == 'nt', reason='POSIX snapshot durability') +@pytest.mark.parametrize('incoming', [False, True]) +def test_snapshotPublicationIsDurable(tmpdir, monkeypatch, incoming): + # A visible rename alone does not make a snapshot survive power loss. + # Both contents and directory entry must be synced before reporting success. + target = str(tmpdir.join('snapshot.dump')) + serializer = Serializer(target, 65536, False, None, None, None) + data = {'marker': 'committed-before-crash'} + events = [] + fsync, rename = os.fsync, os.rename + + def sync(fd): + events.append('directory' if stat.S_ISDIR(os.fstat(fd).st_mode) else 'file') + fsync(fd) + + def replace(old, new): + assert events == ['file'] + rename(old, new) + events.append('rename') + + monkeypatch.setattr(os, 'fsync', sync) + monkeypatch.setattr(os, 'rename', replace) + assert storeSnapshot(serializer, data, incoming) + assert events == ['file', 'rename', 'directory'] + assert serializer.deserialize() == data + + +@pytest.mark.skipif(os.name == 'nt', reason='POSIX snapshot durability') +@pytest.mark.parametrize('mode', ['local', 'fork', 'incoming']) +@pytest.mark.parametrize('failedSync', ['file', 'directory']) +def test_snapshotSyncFailure(tmpdir, monkeypatch, mode, failedSync): + if mode == 'fork' and not hasattr(os, 'fork'): + pytest.skip('fork is not available') + target = str(tmpdir.join('snapshot.dump')) + with open(target, 'wb') as stream: + stream.write(b'previous snapshot') + serializer = Serializer(target, 65536, mode == 'fork', None, None, None) + fsync = os.fsync + + def sync(fd): + kind = 'directory' if stat.S_ISDIR(os.fstat(fd).st_mode) else 'file' + if kind == failedSync: + raise OSError('Injected snapshot sync failure') + fsync(fd) + + monkeypatch.setattr(os, 'fsync', sync) + assert not storeSnapshot(serializer, {'marker': 'new snapshot'}, mode == 'incoming') + if failedSync == 'file': + with open(target, 'rb') as stream: + assert stream.read() == b'previous snapshot' + + def getRandStr(): return '%0100000x' % random.randrange(16 ** 100000)