Setting the file. One moment. File Cu Canary
Previous
Bundled file Document Limits
helpers/_indexer_observation.py
Python·135 lines·7 KB
12
13PROJECTION_FIELDS = {"non_schedule_digest", "schedule_digest", "schedule_raw_digest"}
14TICKS = 10_000_000
15DURATION = re.compile(r"P(?:(?P<days>[0-9]{1,9})D)?(?:T(?:(?P<hours>[0-9]{1,9})H)?"
16 r"(?:(?P<minutes>[0-9]{1,9})M)?(?:(?P<seconds>[0-9]{1,9})(?:\.(?P<fraction>[0-9]{1,7}))?S)?)?")
17INSTANT = re.compile(r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})"
18 r"(?:\.([0-9]{1,7}))?(Z|[+-][0-9]{2}:[0-9]{2})")
19
20
21def note(diagnostics, severity, code, field, message, request_id):
22 entry = dict(severity=severity, code=code, field=field, message=message, request_id=request_id)
23 if entry not in diagnostics:
24 diagnostics.append(entry)
25
26
27def failure(diagnostics, code, field, message, request_id):
28 note(diagnostics, "error", code, field, message, request_id)
29 return HelperFailure(code, message, blocked_at="indexer-verification", request_id=request_id)
30
31
32def interval(value):
33 match = DURATION.fullmatch(value) if isinstance(value, str) and len(value) <= 96 else None
34 if not match or not any(match.group(k) for k in ("days", "hours", "minutes", "seconds")):
35 raise ValueError("Unsupported duration")
36 if "T" in value and not any(match.group(k) for k in ("hours", "minutes", "seconds")):
37 raise ValueError("Empty time component")
38 parts = match.groupdict()
39 seconds = sum(int(parts[k] or 0) * scale for k, scale in
40 (("days", 86400), ("hours", 3600), ("minutes", 60), ("seconds", 1)))
41 ticks = seconds * TICKS + int((parts["fraction"] or "").ljust(7, "0"))
42 if not 300 * TICKS <= ticks <= 86400 * TICKS:
43 raise ValueError("Indexer interval outside documented bounds")
44 return ticks
45
46
47def instant(value):
48 match = INSTANT.fullmatch(value) if isinstance(value, str) and len(value) <= 40 else None
49 if not match or match[8] == "-00:00":
50 raise ValueError("Unknown or unsupported timezone")
51 year, month, day, hour, minute, second = map(int, match.groups()[:6])
52 zone = match[8]
53 offset = 0
54 if zone != "Z":
55 hours, minutes = int(zone[1:3]), int(zone[4:6])
56 if hours > 14 or minutes > 59 or hours == 14 and minutes:
57 raise ValueError("Invalid offset")
58 offset = (hours * 60 + minutes) * (-1 if zone[0] == "-" else 1)
59 stamp = datetime(year, month, day, hour, minute, second,
60 tzinfo=timezone(timedelta(minutes=offset))).astimezone(timezone.utc)
61 elapsed = stamp - datetime(1, 1, 1, tzinfo=timezone.utc)
62 return (elapsed.days * 86400 + elapsed.seconds) * TICKS + int((match[7] or "").ljust(7, "0"))
63
64
65def schedule(value, *, present=True, diagnostics=None, request_id=None):
66 diagnostics = diagnostics if diagnostics is not None else []
67 if value is None:
68 return {"state": "null" if present else "missing"}
69 field = "schedule"
70 try:
71 if not isinstance(value, dict) or set(value) - {"interval", "startTime"}:
72 raise ValueError("Unknown shape")
73 field = "schedule.interval"
74 period = interval(value.get("interval"))
75 field = "schedule.startTime"
76 start = instant(value["startTime"]) if value.get("startTime") is not None else None
77 return {"state": "object", "interval": period, "startTime": start}
78 except (ValueError, OverflowError) as exc:
79 raise failure(diagnostics, "indexer-schedule-evidence-invalid", field,
80 f"Generated indexer {field} is malformed or unsupported; no default is inferred.",
81 request_id) from exc
82
83
84def observe(child, approved, diagnostics, request_id):
85 observed = schedule(child.get("schedule"), present="schedule" in child,
86 diagnostics=diagnostics, request_id=request_id)
87 expected = schedule(approved, diagnostics=diagnostics, request_id=request_id)
88 if approved is not None:
89 for field in ("interval", "startTime"):
90 if field == "startTime" and approved.get(field) is None:
91 continue
92 if observed.get(field) != expected[field]:
93 raise failure(diagnostics, "indexer-schedule-constraint-violation", f"schedule.{field}",
94 f"Generated indexer schedule.{field} violates the approved constraint; this is not an ingestion failure.",
95 request_id)
96 if child["schedule"].get(field) != approved.get(field):
97 note(diagnostics, "info", "indexer-schedule-format-equivalent", f"schedule.{field}",
98 "Schedule formats represent the same duration or instant.", request_id)
99 if observed != expected:
100 note(diagnostics, "warning", "indexer-schedule-unconstrained", "schedule",
101 "Generated schedule differs on unconstrained timing; recurring work/cost is possible. No schedule was changed or default assumed.",
102 request_id)
103 return {
104 "etag": child["@odata.etag"], "digest": digest(child),
105 "non_schedule_digest": digest({k: v for k, v in child.items() if k not in {"schedule", "@odata.etag"}}),
106 "schedule_digest": digest(observed),
107 "schedule_raw_digest": digest({"present": "schedule" in child, "value": child.get("schedule")}),
108 }
109
110
111def compare(current, expected, diagnostics, request_id):
112 if current == expected:
113 return
114 if not PROJECTION_FIELDS <= set(expected):
115 raise failure(diagnostics, "indexer-legacy-evidence-insufficient", "indexer",
116 "Legacy full-hash evidence differs; no schedule preimage/projection exists. Retain it; do not auto-upgrade.",
117 request_id)
118 if current["digest"] == expected["digest"]:
119 raise failure(diagnostics, "indexer-evidence-inconsistent", "indexer",
120 "Equal full hashes have inconsistent version/projection evidence.", request_id)
121 if current["non_schedule_digest"] != expected["non_schedule_digest"]:
122 raise failure(diagnostics, "indexer-definition-drift", "indexer",
123 "A non-schedule indexer field changed; schedule policy cannot admit it.", request_id)
124 if current["schedule_raw_digest"] == expected["schedule_raw_digest"]:
125 raise failure(diagnostics, "indexer-version-unexplained", "@odata.etag",
126 "Indexer version/full hash changed without an observed schedule change; the revision is unproven.", request_id)
127 equivalent = current["schedule_digest"] == expected["schedule_digest"]
128 note(diagnostics, "info" if equivalent else "warning",
129 "indexer-schedule-format-equivalent" if equivalent else "indexer-schedule-unconstrained",
130 "schedule", "Only schedule changed; every other observed field matches the retained projection and approved timing constraints hold.",
131 request_id)
132
133
134def warnings(diagnostics):
135 return list(dict.fromkeys(item["message"] for item in diagnostics if item["severity"] == "warning"))