Reliability: dependencies, restore and recovery evidence

"""Original recovery planning worksheet. No scheduler, database or cloud execution."""
from copy import deepcopy
from graphlib import TopologicalSorter, CycleError
import hashlib
import itertools
import json
from pathlib import Path

GATES = ('data_consistent', 'old_writer_fenced', 'keys_accessible',
 'functional_check', 'business_acceptance')


def minutes(value, name):
 if type(value) is not int or value < 0:
 raise ValueError(name + ' must be a nonnegative integer')
 return value


def assess(tasks, service_task, rto_minutes, recovery_age_minutes, rpo_minutes, checks):
 """All tasks start as soon as dependencies finish, with unlimited workers.

 Durations are fictional estimates from interruption time zero. Checks are
 supplied assertions, not verified evidence. None means unknown. Tasks outside
 the service-task ancestor graph are explicitly listed. RPO uses an externally
 established coherent recovery point; a timestamp alone does not establish one.
 """
 for name, value in [('rto', rto_minutes), ('rpo', rpo_minutes)]:
 minutes(value, name)
 if recovery_age_minutes is not None:
 minutes(recovery_age_minutes, 'recovery age')
 if not isinstance(tasks, list) or not tasks:
 raise ValueError('tasks must be a nonempty list')
 if not isinstance(checks, dict) or set(checks)!= set(GATES):
 raise ValueError('provide exactly the five evidence fields')
 if any(value is not None and type(value) is not bool for value in checks.values):
 raise ValueError('checks must be true, false or null')
 by_id = {}
 for task in tasks:
 if not isinstance(task, dict) or set(task)!= {'id', 'minutes', 'after'}:
 raise ValueError('invalid task fields')
 name = task['id']
 if not isinstance(name, str) or not name.strip or name in by_id:
 raise ValueError('task IDs must be nonempty and unique')
 minutes(task['minutes'], name)
 deps = task['after']
 if not isinstance(deps, list) or any(not isinstance(x, str) for x in deps):
 raise ValueError('dependencies must be strings in a list')
 if len(set(deps))!= len(deps):
 raise ValueError('repeated dependency')
 by_id[name] = task
 if not isinstance(service_task, str) or service_task not in by_id:
 raise ValueError('service task not defined')
 graph = {name: task['after'][:] for name, task in by_id.items}
 if any(dep not in by_id for deps in graph.values for dep in deps):
 raise ValueError('dependency not defined')
 try:
 order = list(TopologicalSorter(graph).static_order)
 except CycleError as error:
 raise ValueError('cyclic recovery plan') from error
 finish, ancestors = {}, {}
 for name in order:
 finish[name] = max((finish[d] for d in graph[name]), default=0) + by_id[name]['minutes']
 ancestors[name] = {name}.union(*(ancestors[d] for d in graph[name]))
 total = finish[service_task]
 rpo_met = None if recovery_age_minutes is None else recovery_age_minutes <= rpo_minutes
 return {
 'estimatedServiceMinutes': total,
 'finishMinutes': dict(sorted(finish.items)),
 'outsideServicePath': sorted(set(graph) - ancestors[service_task]),
 'estimatedRtoMet': total <= rto_minutes,
 'reportedRpoMet': rpo_met,
 'failedChecks': sorted(k for k, v in checks.items if v is False),
 'unknownChecks': sorted(k for k, v in checks.items if v is None),
 'readyForDecision': total <= rto_minutes and rpo_met is True and all(v is True for v in checks.values),
 'productionAuthorized': False,
 }


def task(name, duration, *after):
 return {'id': name, 'minutes': duration, 'after': list(after)}


def evidence:
 tasks = [task('cluster', 12), task('keys', 3), task('database', 18),
 task('fence', 5), task('workload', 8, 'cluster', 'keys'),
 task('validate', 6, 'database', 'workload', 'fence'),
 task('service', 4, 'validate')]
 good = dict.fromkeys(GATES, True)
 base = [tasks, 'service', 30, 4, 5, good]
 original = deepcopy(base)
 fixtures = []
 for name, change in [
 ('parallel-plan', {}), ('rto-missed', {2: 29}),
 ('rpo-missed', {3: 6}), ('unknown-recovery-point', {3: None}),
 ('unknown-acceptance', {5: {**good, 'business_acceptance': None}}),
 ('old-writer-active', {5: {**good, 'old_writer_fenced': False}}),
 ('cleanup-outside-service', {0: tasks + [task('cleanup', 60, 'service')]}),
 ('slow-cluster', {0: [{**t, 'minutes': 17} if t['id'] == 'cluster' else t for t in tasks]}),
 ]:
 args = deepcopy(base)
 for index, value in change.items:
 args[index] = value
 fixtures.append({'id': name, **assess(*args)})
 assert fixtures[0]['estimatedServiceMinutes'] == 30
 assert fixtures[0]['readyForDecision'] is True
 assert all(not f['productionAuthorized'] for f in fixtures)
 assert all(not f['readyForDecision'] for f in fixtures[1:6])
 assert fixtures[6]['estimatedServiceMinutes'] == 30
 assert fixtures[6]['outsideServicePath'] == ['cleanup']
 assert fixtures[7]['estimatedServiceMinutes'] == 35
 duration_checks = 0
 for a, b, c in itertools.product(range(8), repeat=3):
 model = [task('a', a), task('b', b), task('c', c, 'a', 'b')]
 result = assess(model, 'c', 100, 0, 0, good)
 assert result['estimatedServiceMinutes'] == max(a, b) + c
 assert assess(list(reversed(model)), 'c', 100, 0, 0, good) == result
 duration_checks += 1
 gate_checks = 0
 for values in itertools.product((True, False, None), repeat=5):
 checks = dict(zip(GATES, values))
 result = assess(tasks, 'service', 30, 4, 5, checks)
 assert result['readyForDecision'] == all(v is True for v in values)
 assert result['unknownChecks'] == sorted(k for k,v in checks.items if v is None)
 assert result['failedChecks'] == sorted(k for k,v in checks.items if v is False)
 gate_checks += 1
 invalid = [
 {0: []}, {0: tasks + [tasks[0]]}, {0: [task('a', -1)]},
 {0: [task('a', True)]}, {0: [task('service', 1, 'missing')]},
 {0: [task('service', 1, 'x'), task('x', 1, 'service')]},
 {0: [task('service', 1, 'service')]}, {1: 'absent'},
 {2: True}, {3: -1}, {4: 1.2}, {5: {}},
 {5: {**good, 'functional_check': 1}},
 {0: [task('a', 1), task('service', 2, 'a', 'a')]},
 ]
 for change in invalid:
 args = deepcopy(base)
 for index, value in change.items:
 args[index] = value
 try:
 assess(*args)
 except ValueError:
 pass
 else:
 raise AssertionError('invalid input accepted')
 assert base == original
 return {'fixtures': fixtures, 'durationCombinations': duration_checks,
 'gateCombinations': gate_checks, 'invalidInputs': len(invalid),
 'inputPreserved': True, 'cloudExecuted': False, 'databaseRestored': False,
 'network': False, 'persistentWrites': False, 'independentVerification': False,
 'scriptSha256': hashlib.sha256(Path(__file__).read_bytes).hexdigest}


if __name__ == '__main__':
 print(json.dumps(evidence, sort_keys=True, indent=2))

Related topics: Processing continuity and reconciliation · Identity and key management · Incident communication and shift handover

Reference: Disaster recovery planning guide · Current linked guide; edition date unconfirmed (2026-09-30 inspection)