Setting the file. One moment. File Cu Canary
— line 234
This file
- Number
- 10.32
- Position
- 32 of 77
- Type
- Python
- Size
- 12 KB
- Lines
- 259
helpers/_progress.py
Python·259 lines·12 KB
sys
12import time
13import threading
14
15try:
16 from ._common import HelperFailure
17except ImportError:
18 from _common import HelperFailure
19
20
21STAGES = {
22 "file-source": ("validation", "source-reconciliation", "file-inventory", "file-upload", "file-readback"),
23 "file-upload": ("file-inventory", "file-upload", "file-readback"),
24 "blob-source": ("validation", "blob-inventory", "source-reconciliation", "ingestion-cycle",
25 "blob-readback", "source-readback"),
26 "blob-monitor": ("ingestion-cycle",),
27 "blob-capture": ("evidence-validation", "context-check", "source-binding", "blob-inventory", "checkpoint"),
28 "blob-recheck": ("evidence-validation", "context-check", "source-binding", "blob-inventory",
29 "ingestion-cycle", "blob-readback", "source-readback", "context-readback"),
30 "search-bootstrap": ("validation", "context-check", "region-check", "absence-check", "search-submit",
31 "arm-wait", "arm-readback"),
32}
33COUNTS = frozenset(("uploads_acknowledged", "files_reused", "files_verified",
34 "status_checks", "cycle_updates_processed", "cycle_items_skipped",
35 "files_failed", "files_unverified", "files_not_attempted", "files_pending", "files_ingested"))
36FILE_HEARTBEAT_SECONDS = 5
37IO_WARNING = "progress-output-failed: stderr progress could not be written; execution result is authoritative."
38SHUTDOWN_WARNING = "progress-shutdown-flush-unresolved: failed stderr could not be redirected; shutdown may override the exit code."
39
40
41def validate_blob_progress(value):
42 fields = {"schema_version", "phase", "run_start", "processed", "failed", "skipped",
43 "unit", "total", "remaining", "denominator", "synchronization_status",
44 "elapsed_seconds", "next_check_seconds"}
45 if not isinstance(value, dict) or set(value) != fields:
46 raise ValueError("Invalid Blob progress shape.")
47 if (value["schema_version"] != "1.0" or value["unit"] != "item-updates"
48 or value["denominator"] != "not-comparable-to-files"
49 or value["total"] is not None or value["remaining"] is not None
50 or value["phase"] not in ("ingesting", "waiting", "throttled", "paused", "completed", "failed")
51 or value["synchronization_status"] not in ("not-reported", "active", "creating", "deleting")):
52 raise ValueError("Invalid Blob progress labels.")
53 for field in ("processed", "failed", "skipped"):
54 count = value[field]
55 if count is not None and (type(count) is not int or count < 0):
56 raise ValueError("Invalid Blob progress count.")
57 for field in ("elapsed_seconds", "next_check_seconds"):
58 number = value[field]
59 if field == "next_check_seconds" and number is None:
60 continue
61 if type(number) not in (int, float) or not math.isfinite(number) or number < 0:
62 raise ValueError("Invalid Blob progress timing.")
63 if value["next_check_seconds"] is not None and value["next_check_seconds"] > 60:
64 raise ValueError("Invalid Blob progress wait.")
65 start = value["run_start"]
66 if start is not None:
67 if not isinstance(start, str) or len(start) > 40:
68 raise ValueError("Invalid Blob progress run.")
69 parsed = datetime.fromisoformat(start.replace("Z", "+00:00"))
70 if parsed.tzinfo is None or parsed.isoformat() != start:
71 raise ValueError("Blob progress run must be a canonical timestamp.")
72
73
74class Progress:
75 def __init__(self, workflow, *, enabled=True, stream=None, clock=time.monotonic):
76 self.stages = STAGES[workflow]
77 self.workflow = workflow
78 self.enabled = enabled
79 self.stream = stream
80 self.clock = clock
81 self.depth = 0
82 self.stage = self.stages[0]
83 self.counts = {}
84 self.started = None
85 self.elapsed = 0.0
86 self.last_sent = None
87 self.last_stage = None
88 self.output_failed = False
89 self.shutdown_flush_unresolved = False
90 self.blob_progress = None
91 self.last_blob_phase = None
92
93 def blob_update(self, value):
94 if not self.workflow.startswith("blob-"):
95 raise ValueError("Blob progress cannot describe a File or ARM workflow.")
96 validate_blob_progress(value)
97 self.blob_progress = copy.deepcopy(value)
98 for field, target in (("processed", "cycle_updates_processed"), ("skipped", "cycle_items_skipped")):
99 self.counts.pop(target, None)
100 if value[field] is not None:
101 self.counts[target] = value[field]
102 self.update("ingestion-cycle")
103
104 def update(self, stage, **counts):
105 if stage not in self.stages or self.stages.index(stage) < self.stages.index(self.stage):
106 raise ValueError("Invalid progress stage transition.")
107 if any(key not in COUNTS or type(value) is not int or value < 0 for key, value in counts.items()):
108 raise ValueError("Invalid progress count.")
109 self.stage = stage
110 self.counts.update(counts)
111 self._emit("running")
112
113 def waiting(self, seconds):
114 if not isinstance(seconds, (int, float)) or not math.isfinite(seconds) or not 0 <= seconds <= 30:
115 raise ValueError("Invalid recovery wait.")
116 self._emit("waiting", wait={"reason": "http-429", "seconds": seconds})
117
118 @contextmanager
119 def processing_file(self, ordinal, total, attempt):
120 if (self.workflow not in {"file-source", "file-upload"} or type(ordinal) is not int
121 or type(total) is not int or not 1 <= ordinal <= total <= 200 or attempt not in (1, 2)):
122 raise ValueError("Invalid active file.")
123 active = {"ordinal": ordinal, "total": total, "attempt": attempt}
124 if not self.enabled or self.output_failed:
125 yield
126 return
127 started = self.clock()
128 stop = threading.Event()
129
130 def pulse():
131 while not stop.wait(FILE_HEARTBEAT_SECONDS):
132 if self.output_failed:
133 return
134 self._emit("processing", active=active, active_started=started)
135
136 self._emit("processing", active=active, active_started=started)
137 worker = threading.Thread(target=pulse, name="foundry-file-progress", daemon=True)
138 worker.start()
139 try:
140 yield
141 finally:
142 stop.set()
143 worker.join()
144
145 def _emit(self, state, *, wait=None, active=None, active_started=None):
146 if not self.enabled or self.output_failed:
147 return
148 observed = self.clock()
149 # Clock regressions/nonfinite observations must not create negative time
150 # or bypass throttling. Progress never shares the service deadline clock.
151 if math.isfinite(observed):
152 if self.started is None:
153 self.started = observed
154 delta = observed - self.started
155 if math.isfinite(delta):
156 self.elapsed = max(self.elapsed, delta)
157 phase = self.blob_progress["phase"] if self.blob_progress is not None else None
158 if (state == "running" and self.stage == self.last_stage and phase == self.last_blob_phase
159 and self.last_sent is not None and self.elapsed - self.last_sent < 1.0):
160 return
161 event = {
162 "event": "progress", "workflow": self.workflow, "activity": self.stage,
163 "state": state, "elapsed_seconds": round(self.elapsed, 3),
164 "remaining_checks": [] if state == "completed" else list(self.stages[self.stages.index(self.stage) + 1:]),
165 }
166 if self.counts:
167 event["completed_counts"] = dict(self.counts)
168 if wait is not None:
169 event["wait"] = wait
170 if active is not None:
171 duration = observed - active_started
172 event["processing_file"] = {**active, "elapsed_seconds": round(max(0, duration), 3) if math.isfinite(duration) else 0}
173 if self.blob_progress is not None:
174 value = self.blob_progress
175 event["blob_progress"] = copy.deepcopy(value)
176 counts = "; ".join(f"{field}={value[field] if value[field] is not None else 'unknown'}"
177 for field in ("processed", "failed", "skipped"))
178 next_check = ("paused/resumable" if value["phase"] == "paused" else "none") if value["next_check_seconds"] is None else f"{value['next_check_seconds']:g}s"
179 event["message"] = (
180 f"Blob ingestion observation {value['phase']}: item updates {counts}; file total/remaining unknown "
181 f"(not comparable); elapsed={value['elapsed_seconds']:g}s; next check={next_check}."
182 )
183 stream = self.stream if self.stream is not None else sys.stderr
184 try:
185 text = json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n"
186 if stream.write(text) != len(text):
187 raise OSError("Short progress write.")
188 stream.flush()
189 except (OSError, UnicodeError, ValueError):
190 # Continue the approved operation, retaining a fixed secondary warning
191 # even when stderr itself is broken. Never replace a primary failure.
192 self.output_failed = True
193 self._disable_failed_default_stderr(stream)
194 self.last_sent, self.last_stage = self.elapsed, self.stage
195 self.last_blob_phase = phase
196
197 def _disable_failed_default_stderr(self, stream):
198 if self.stream is not None or stream is not sys.__stderr__ or stream.closed:
199 return
200 try:
201 if stream.fileno() == 2:
202 # A failed TextIOWrapper flush can retain pending bytes. Redirect
203 # only the failed native stderr so shutdown can drain that buffer
204 # without replacing the authoritative process exit status.
205 sink = os.open(os.devnull, os.O_WRONLY)
206 try:
207 os.dup2(sink, 2)
208 finally:
209 if sink != 2:
210 os.close(sink)
211 except (OSError, ValueError):
212 self.shutdown_flush_unresolved = True
213
214 def finish(self, result=None, failure=None):
215 if self.blob_progress is not None and failure is not None:
216 self.blob_progress.update(phase="failed", next_check_seconds=None)
217 if failure is not None:
218 state = "partial" if failure.partial or failure.writes else "blocked"
219 else:
220 state = {"verified": "completed", "unverified": "blocked"}.get(result["status"], result["status"])
221 if state not in {"completed", "blocked", "partial"}:
222 raise ValueError("Invalid progress terminal state.")
223 self._emit(state)
224 if self.output_failed:
225 warnings = failure.warnings if failure is not None else result.setdefault("warnings", [])
226 if IO_WARNING not in warnings:
227 warnings.append(IO_WARNING)
228 if self.shutdown_flush_unresolved and SHUTDOWN_WARNING not in warnings:
229 warnings.append(SHUTDOWN_WARNING)
230
231
232def reporting(workflow):
233 """One reporter and terminal event across nested public entrypoints."""
234 def decorate(function):
235 @wraps(function)
236 def wrapped(*args, progress=None, **kwargs):
237 if progress is None:
238 progress = Progress(workflow, enabled=False)
239 outer = progress.depth == 0
240 progress.depth += 1
241 try:
242 result = function(*args, progress=progress, **kwargs)
243 except HelperFailure as failure:
244 if outer:
245 progress.finish(failure=failure)
246 raise
247 else:
248 if outer:
249 progress.finish(result=result)
250 return result
251 finally:
252 progress.depth -= 1
253 return wrapped
254 return decorate
255
256
257def add_progress_argument(parser):
258 parser.add_argument("--no-progress", dest="progress", action="store_false",
259 help="Suppress content-free execution progress on stderr.")