Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576"""Real-process contract for Jetstream V2 planSnapshot operator controls."""
import jsonimport osimport urllib.errorimport urllib.request
BASE = os.environ["STREAM_BASE_URL"].rstrip("/") + "/xrpc/network.bsky.jetstream.planSnapshot"MODE = os.environ["STREAM_PLAN_CONFIG_MODE"]
# listener-first startup: xrpc paths 503 as "starting" until storage is# published; wait out the gate before asserting plan responsesimport time as _time
_status_url = os.environ["STREAM_BASE_URL"].rstrip("/") + "/status"_deadline = _time.monotonic() + 15while True: try: with urllib.request.urlopen(_status_url, timeout=1) as _r: if b"state starting" not in _r.read(): break except OSError: pass assert _time.monotonic() < _deadline, "startup gate never opened" _time.sleep(0.05)
def post(body): request = urllib.request.Request( BASE, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST", ) try: response = urllib.request.urlopen(request) return response.status, json.load(response) except urllib.error.HTTPError as error: return error.code, json.load(error)
if MODE == "limited": status, result = post({"collections": ["app.bsky.feed.post"]}) assert status == 200 assert result["sealedTipSeq"] == 12 assert result["plannedThroughSeq"] == 1 assert result["stats"]["entries"] == 1 assert len(result["segments"]) == 1 assert result["segments"][0]["mode"] == "blocks" assert result["segments"][0]["blocks"] == [{"first": 0, "last": 0}] for body in ( {"dids": ["did:plc:a", "did:plc:b"]}, {"collections": ["app.bsky.feed.post", "app.bsky.feed.like"]}, ): status, error = post(body) assert status == 400 and error["error"] == "InvalidRequest"elif MODE == "whole": status, result = post({"collections": ["app.bsky.feed.post"]}) assert status == 200 assert result["plannedThroughSeq"] == result["sealedTipSeq"] == 12 assert result["stats"]["entries"] == 3 assert len(result["segments"]) == 3 assert all(segment["mode"] == "segment" for segment in result["segments"])elif MODE == "disabled": for body in ( {"dids": ["did:plc:planconfigaccount"]}, {"collections": ["app.bsky.feed.post"]}, ): status, error = post(body) assert status == 400 and error["error"] == "InvalidRequest"else: raise AssertionError(f"unknown mode: {MODE}")
print(f"plan config {MODE}: PASS")