zc2 0.0.31

P2P compute broker with credit-based billing, WAL, and broker mesh support
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

#!/usr/bin/env python3
"""
Publish 10 fibonacci calculation jobs to the broker /execute endpoint.
Verifies that workers are assigned and return the expected results.

Requires: ZAKURO_API_KEY (and ZAKURO_API_URL, default http://localhost:9000).
Usage:
  export ZAKURO_API_KEY=zk_xxx
  python3 scripts/trigger_10_fibonacci_jobs.py

With broker and fibonacci worker running (e.g. via scripts/e2e_fibonacci_broadcast_30s.sh
without triggering the 30s task), this sends 10 quick jobs and prints each result.
"""

import json
import os
import sys
import time
import urllib.request
import urllib.error

# fibonacci(4) = 3, fibonacci(10) = 55
EXPECTED = {4: 3, 10: 55}


def send_one(base: str, key: str, job_id: int, n: int = 4) -> tuple[bool, str, float]:
    """Send one /execute request. Returns (ok, message, elapsed_sec)."""
    url = f"{base.rstrip('/')}/execute"
    body = json.dumps({"n": n}).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {key}",
            "X-Zakuro-Requirements": json.dumps({
                "strategy": "round_robin",
                "estimated_duration_secs": 1,
            }),
        },
    )
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = resp.read()
            elapsed = time.perf_counter() - t0
            # Broker may return application/octet-stream with worker's JSON body
            try:
                text = data.decode("utf-8", errors="replace")
                j = json.loads(text)
                result = j.get("result")
                expected = EXPECTED.get(n)
                ok = result == expected if expected is not None else result is not None
                msg = f"fib({n})={result}" + (f" (expected {expected})" if expected is not None else "")
                return ok, msg, elapsed
            except json.JSONDecodeError as e:
                return False, f"not JSON: {text[:80]!r}...", elapsed
    except urllib.error.HTTPError as e:
        elapsed = time.perf_counter() - t0
        body = e.read()[:500]
        try:
            body_str = body.decode("utf-8", errors="replace").strip()
        except Exception:
            body_str = repr(body)
        body_short = body_str[:60] + "..." if len(body_str) > 60 else body_str
        return False, f"HTTP {e.code} {e.reason}: {body_short}", elapsed
    except Exception as e:
        elapsed = time.perf_counter() - t0
        return False, str(e), elapsed


def main() -> None:
    base = os.environ.get("ZAKURO_API_URL", "http://localhost:9000")
    key = os.environ.get("ZAKURO_API_KEY")
    if not key:
        print("Set ZAKURO_API_KEY to run this script.", file=sys.stderr)
        print("Example: export ZAKURO_API_KEY=zk_1000000001_xxx", file=sys.stderr)
        sys.exit(1)

    n_per_job = 4  # fib(4) = 3
    total = 10
    print(f"Publishing {total} fibonacci jobs (n={n_per_job}, expected result={EXPECTED.get(n_per_job)}) to {base}/execute ...")
    print()

    ok_count = 0
    for i in range(total):
        ok, msg, elapsed = send_one(base, key, i + 1, n=n_per_job)
        if ok:
            ok_count += 1
            print(f"  Job {i+1:2d}/{total}: OK  {msg}  ({elapsed:.2f}s)")
        else:
            print(f"  Job {i+1:2d}/{total}: FAIL  {msg}  ({elapsed:.2f}s)")

    print()
    if ok_count == total:
        print(f"All {total} jobs completed successfully. Workers were assigned and returned correct results.")
        sys.exit(0)
    else:
        print(f"Only {ok_count}/{total} jobs succeeded.")
        print("Ensure broker and fibonacci worker are running and ZAKURO_API_KEY is valid.")
        sys.exit(1)


if __name__ == "__main__":
    main()