Continuidade de serviço: recuperar trabalho e confirmar resultados

"""Original teaching model. No Pub/Sub client, real-time scheduler or cloud calls."""
import hashlib
import itertools
import json
from pathlib import Path


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


def model(initial, phases, quarantined=0):
    """Each second admits arrivals, then completes min(queue, worker, downstream).
    Initial quarantined work leaves the active queue but is never completed.
    All units are equally costly. No failures, retries, latency or expiry.
    """
    integer(initial, 'initial')
    integer(quarantined, 'quarantined')
    if quarantined > initial:
        raise ValueError('quarantine exceeds initial work')
    if not isinstance(phases, list):
        raise ValueError('phases must be a list')
    checked = []
    for phase in phases:
        if not isinstance(phase, dict) or set(phase) != {'seconds', 'arrivals', 'worker', 'downstream'}:
            raise ValueError('phase fields must match the model')
        checked.append({k: integer(v, k) for k, v in phase.items()})
    queue, accepted, completed, elapsed = initial - quarantined, initial, 0, 0
    first_empty = 0 if queue == 0 else None
    points = []
    for phase in checked:
        capacity = min(phase['worker'], phase['downstream'])
        for _ in range(phase['seconds']):
            queue += phase['arrivals']
            accepted += phase['arrivals']
            done = min(queue, capacity)
            queue -= done
            completed += done
            elapsed += 1
            if queue == 0 and first_empty is None:
                first_empty = elapsed
            assert accepted == completed + queue + quarantined
        points.append({'second': elapsed, 'pending': queue, 'completed': completed})
    return {'accepted': accepted, 'completed': completed, 'pending': queue,
            'quarantined': quarantined, 'firstEmptySecond': first_empty,
            'allAcceptedCompleted': queue == 0 and quarantined == 0,
            'phaseEnds': points}


def phase(seconds, arrivals, worker, downstream):
    return dict(seconds=seconds, arrivals=arrivals, worker=worker, downstream=downstream)


def main():
    scenarios = [
        ('net-capacity', 12000, [phase(300, 80, 120, 120)], 0, 0, 300),
        ('balanced', 4000, [phase(300, 90, 90, 90)], 0, 4000, None),
        ('downstream-cap', 9000, [phase(240, 80, 200, 110)], 0, 1800, None),
        ('downstream-drained', 9000, [phase(300, 80, 200, 110)], 0, 0, 300),
        ('load-change', 1000, [phase(10, 20, 60, 60), phase(10, 70, 60, 60)], 0, 700, None),
        ('quarantine', 1000, [phase(10, 0, 100, 100)], 20, 0, 10),
        ('empty-then-regrows', 10, [phase(1, 0, 10, 10), phase(2, 20, 10, 10)], 0, 20, 1),
        ('no-capacity', 4, [phase(3, 2, 0, 10)], 0, 10, None),
    ]
    fixtures = []
    for name, initial, phases, quarantine, pending, first_empty in scenarios:
        before = json.dumps(phases, sort_keys=True)
        result = model(initial, phases, quarantine)
        assert result['pending'] == pending and result['firstEmptySecond'] == first_empty
        assert json.dumps(phases, sort_keys=True) == before
        fixtures.append({'id': name, **result})
    assert fixtures[5]['completed'] == 980 and not fixtures[5]['allAcceptedCompleted']
    assert fixtures[6]['firstEmptySecond'] == 1 and fixtures[6]['pending'] == 20
    combinations = 0
    for initial, arrivals, worker, downstream, seconds in itertools.product(range(5), range(4), range(4), range(4), range(5)):
        result = model(initial, [phase(seconds, arrivals, worker, downstream)])
        expected = max(0, initial + seconds * (arrivals - min(worker, downstream)))
        assert result['pending'] == expected
        assert result['accepted'] == initial + seconds * arrivals
        assert result['completed'] == result['accepted'] - expected
        combinations += 1
    invalid = [(-1, [], 0), (True, [], 0), (1.5, [], 0), (1, [], -1), (1, [], 2),
               (1, {}, 0), (1, [{}], 0), (1, [phase(1, -1, 1, 1)], 0),
               (1, [phase(True, 1, 1, 1)], 0), (1, [phase(1, 1, '2', 1)], 0)]
    for args in invalid:
        try:
            model(*args)
        except ValueError:
            pass
        else:
            raise AssertionError('invalid input accepted')
    print(json.dumps({'scriptSha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        'fixtures': fixtures, 'constantRateCombinations': combinations, 'invalidInputs': len(invalid),
        'inputPreserved': True, 'cloudExecuted': False, 'network': False, 'persistentWrites': False,
        'limitations': 'Integer equal-cost work model; no vendor execution, deadlines, retries, expiry, per-message order or delivery guarantees.'}, indent=2))


if __name__ == '__main__':
    main()

Tópicos relacionados: SLOs e capacidade de serviço · Recuperação de hooks e identidade de intenção · Telemetria durante rollout

Referência: Professional Cloud DevOps Engineer exam guide · Current linked guide; edition date unconfirmed (2026-09-30 inspection)