chronaxis.results.raw_exports
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""保存原始片段索引并复制已生成的BLF，不重新调用旧读写或解码路径。23输入是实际Segment范围、已关闭的BLF文件及其明确摘要，输出是连续编号的有界4JSONL索引和发布目录内的逐文件副本。exported保留正长actual，即使文件含零帧；5empty_range保留已知零长度actual；no_observation仅用于actual为None，不造空文件。6append核实际BLF字节/SHA后写索引；seal真正关闭writer才开放独立reader。reader只有7完整EOF、流SHA/数量与真实关闭均成功后才verified_eof，提前关闭不能证明完整。89所有新增RAM与索引spool先预约，原BLF由旧writer封装预约，本模块不重复计算。10copy_to只逐条读索引、有限块复制和核现成BLF；发布副本不计入分析暂存配额。11输出失败只清本次复制资源，保sealed输入与其他reader，恢复后可向新目录重试；12原索引或原BLF内容损坏仍使输入仓failed，不把坏输入当作可发布结果。13所有真实FD、未返回reader和费用槽由本store可达，关闭失败可重试，保原异常/取消14及最高raw_export_store_cleanup_owner；文件与累计spool保留，不凭共享余额补账。15本模块只依赖标准库、冻结模型和有界JSON，不导入CAN库，不声称整进程RSS受限。16"""1718# 固定模型与有界编码不触及原日志/DBC；资源交易复用新结果层已经验证的协议。19from dataclasses import dataclass20from hashlib import sha25621from pathlib import Path22from re import fullmatch23from ..core.models import Interval24from ..core.validation import require_identifier25from ..legacy.records import MAX_WIRE_BYTES, encode_packet, decode_packet26from ..timebase.conversion import INT64_MAX, require_int64_ns27from .budget import ResourceBudget28from .errors import ResultClosedError, ResultFormatError29from .spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram303132_FIELDS = ('index', 'source', 'session_id', 'segment_id', 'window_id', 'requested',33 'actual', 'frame_count', 'file_name', 'sha256', 'byte_count', 'status', 'device_label')34_ROW_COST = 65536 + 64 * MAX_WIRE_BYTES35_CHUNK_BYTES = 6553636_IO_COST = 8 * _CHUNK_BYTES37_MAX_INDEX = 999999999999383940def _integer(value, name, maximum=INT64_MAX):41 """校验非负原生整数，不把bool或浮点作为数量。4243 参数: value候选；name错误位置；maximum含端点上限。返回: 原int。44 异常: TypeError/ValueError为类型或范围错误；不截断、不转换。45 """46 if type(value) is not int:47 raise TypeError(name + '需要原生int')48 if not 0 <= value <= maximum:49 raise ValueError(name + '超出允许范围')50 return value515253def _text(value, name, *, identifier=False):54 """检查有界安全原文本，保留大小写、Unicode和合法ID标点。5556 参数: value原文本；name位置；identifier要求来源/会话简单名。返回: 原str。57 异常: TypeError/ValueError为非原生str、空白、控制字符、过长或名称非法。58 """59 if type(value) is not str:60 raise TypeError(name + '需要原生str')61 if not value.strip() or len(value) > MAX_WIRE_BYTES or any(ord(char) < 32 or ord(char) == 127 for char in value):62 raise ValueError(name + '为空、含控制字符或超过有界文本能力')63 if identifier:64 require_identifier(value, name)65 return value666768def _interval(value, name):69 """要求冻结Interval及原生int64端点，不重建或改变原范围。7071 参数: value范围；name位置。返回: 原Interval。72 异常: TypeError/ValueError/OverflowError为类型、顺序或整数范围非法。73 """74 if type(value) is not Interval:75 raise TypeError(name + '需要Interval')76 for endpoint in (value.start_ns, value.end_ns):77 if type(endpoint) is not int:78 raise TypeError(name + '端点需要原生int64')79 require_int64_ns(endpoint, name)80 require_int64_ns(value.end_ns - value.start_ns, name + '.duration')81 if value.end_ns < value.start_ns:82 raise ValueError(name + '不能反向')83 return value848586def _packet(record):87 """把固定模型转为固定JSON树，不复制文件内容或任意元数据。8889 参数: record为RawExportRecord。返回: 仅13字段dict。90 异常: 内部模型属性错误传播；调用者另核整包64KiB上限。91 """92 result = {name: getattr(record, name) for name in _FIELDS}93 for name in ('requested', 'actual'):94 interval = result[name]95 result[name] = None if interval is None else dict(start_ns=interval.start_ns, end_ns=interval.end_ns)96 return result979899@dataclass(frozen=True)100class RawExportRecord:101 """保存一个原片段导出事实；构造校声明，不据此宣称文件已读取或写出。"""102103 index: int # 从0连续；对应唯一12位文件名，非业务排序键。104 source: str # 同一store的CAN来源安全名。105 session_id: str # 原会话安全名。106 segment_id: str # 实际Segment的原ID。107 window_id: str # 产生原片段的窗口ID。108 requested: Interval # 保留完整原请求范围。109 actual: Interval | None # 原实际交集；已知点与无观测严格区分。110 frame_count: int # 原writer成功交付的帧数；正长范围可为0。111 file_name: str | None # exported固定为raw_segments/12位.blf。112 sha256: str | None # 实际输出BLF字节SHA，不是原输入SHA。113 byte_count: int # 实际输出文件字节数。114 status: str # exported/empty_range/no_observation。115 device_label: str # 本条产生设备标签，与store严格一致。116117 def __post_init__(self):118 """检查数量、身份、范围和文件状态的闭合关系，整条记录有界。119120 参数: 无，读取冻结字段。返回: None。121 异常: TypeError/ValueError/OverflowError为非法事实；不打开、stat或创建文件。122 """123 _integer(self.index, 'raw_export.index', _MAX_INDEX)124 _integer(self.frame_count, 'raw_export.frame_count')125 _integer(self.byte_count, 'raw_export.byte_count')126 for name in ('source', 'session_id', 'segment_id', 'window_id', 'device_label'):127 _text(getattr(self, name), name, identifier=name in ('source', 'session_id'))128 _interval(self.requested, 'raw_export.requested')129 if self.actual is not None:130 _interval(self.actual, 'raw_export.actual')131 if self.actual.start_ns < self.requested.start_ns or self.actual.end_ns > self.requested.end_ns:132 raise ValueError('raw_export.actual必须落在原requested内')133 if type(self.status) is not str or self.status not in ('exported', 'empty_range', 'no_observation'):134 raise ValueError('原始导出状态非法')135 if self.status == 'exported':136 if self.actual is None or self.actual.start_ns == self.actual.end_ns:137 raise ValueError('exported需正长实际范围')138 if type(self.file_name) is not str or self.file_name != f'raw_segments/{self.index:012d}.blf':139 raise ValueError('原始导出文件名必须对应当前连续索引')140 if type(self.sha256) is not str or fullmatch(r'[0-9a-fA-F]{64}', self.sha256) is None:141 raise ValueError('原始导出SHA需64位十六进制')142 else:143 expected = self.actual is None if self.status == 'no_observation' else self.actual is not None and self.actual.start_ns == self.actual.end_ns144 if not expected or self.file_name is not None or self.sha256 is not None or self.frame_count or self.byte_count:145 raise ValueError('无观测/零范围必须保原actual且不得带文件、SHA或帧字节数')146 # 在模型边界按现有wire能力校整包，避免多个单独合法长文本拼成无界行。147 if len(encode_packet(_packet(self))) + 1 > MAX_WIRE_BYTES:148 raise ValueError('原始导出索引行超过64KiB')149150151def _decode(payload):152 """严格还原一条固定索引，不接受多键、重复JSON键或浮点端点。153154 参数: payload为不含换行的有限bytes。返回: RawExportRecord。155 异常: ResultFormatError保留底层结构/类型/时间校验原因；取消原样传播。156 """157 try:158 value = decode_packet(payload)159 if set(value) != set(_FIELDS):160 raise ValueError('原始导出索引字段缺失或多余')161 for name in ('requested', 'actual'):162 interval = value[name]163 if interval is None and name == 'actual':164 continue165 if type(interval) is not dict or set(interval) != {'start_ns', 'end_ns'}:166 raise ValueError('索引范围必须精确含start_ns/end_ns')167 if type(interval['start_ns']) is not int or type(interval['end_ns']) is not int:168 raise TypeError('索引范围端点必须为原生int')169 value[name] = Interval(interval['start_ns'], interval['end_ns'])170 return RawExportRecord(**value)171 except (TypeError, ValueError, OverflowError, UnicodeError) as exc:172 raise ResultFormatError('原始导出索引行非法: ' + str(exc)) from exc173174175def _write_all(handle, payload):176 """完整写有限bytes，短写继续，零写或非法返回立即失败。177178 参数: handle真实无缓冲FD；payload有限bytes。返回: None。179 异常: 原write/取消或OSError；费用由调用方先预约，不在此重复计算。180 """181 view = memoryview(payload)182 while view:183 written = handle.write(view)184 if type(written) is not int or not 0 < written <= len(view):185 raise OSError('原始导出文件写入未正确推进')186 view = view[written:]187188189class RawExportStore:190 """拥有索引、活reader与当前复制FD；状态与真实closed事实分别记录。191192 building只可append/seal；sealed可独立回读或重复copy。构建或输入完整性失败193 转failed；普通复制输出失败仅恢复本次资源。关闭中只允许重试close，不丢槽。194 """195196 format = 'blf'197198 def __init__(self, budget, work_dir, source, session_id, device_label='pending'):199 """先付固定费用，再在全新根独占创建目录和索引writer。200201 参数: budget为借用ResourceBudget；work_dir为全新str/Path；source/session_id202 为安全名；device_label为有界原标签，默认pending。203 返回: building仓。异常: 原声明/路径/预算/IO/取消；持续清理失败附最高恢复owner。204 """205 if not isinstance(budget, ResourceBudget) or not isinstance(work_dir, (str, Path)):206 raise TypeError('原始导出仓需ResourceBudget及显式新目录')207 _text(source, 'raw_exports.source', identifier=True)208 _text(session_id, 'raw_exports.session_id', identifier=True)209 _text(device_label, 'raw_exports.device_label')210 if sum(len(part) for part in Path(work_dir).parts) > MAX_WIRE_BYTES:211 raise ValueError('原始导出工作路径超过有界能力')212 self.budget = self._budget = budget213 self.source, self.session_id, self.device_label = source, session_id, device_label214 self.path = self.inventory_path = self._raw_path = None215 self._writer = self._pending_reader = self._copy_reader = None216 self._input_fd = self._output_fd = self._inventory_fd = None217 self._readers = self._digest = self._payload = self._chunk = None218 self._fixed_cost = self._append_cost = self._io_cost = self._copy_cost = self._held_cost = 0219 self._spool_bytes = self.inventory_bytes = self.record_count = self.file_count = 0220 self.sha256 = None221 self.state, self._closed, self._closing = 'building', False, False222 try:223 # 路径/来源标签及活reader登记均有固定费用，首笔后取消也可从完整槽恢复。224 requested = Path(work_dir)225 _reserve_owned_ram(self, 65536 + 32 * (len(str(requested)) + len(source) + len(session_id) + len(device_label)), '_fixed_cost')226 resolved = requested.resolve()227 resolved.mkdir(parents=True, exist_ok=False)228 self.path, self._raw_path = resolved, resolved / 'raw_segments'229 self._raw_path.mkdir(exist_ok=False)230 self.inventory_path = resolved / 'raw_segments.jsonl'231 self._readers, self._digest = set(), sha256()232 self._writer = self.inventory_path.open('xb', buffering=0)233 except BaseException as exc:234 self._fail(exc)235 raise236237 @property238 def closed(self):239 """读取真实关闭事实。参数: 无。返回: bool。异常: 无，FD或费用未清不为True。"""240 return self._closed241242 @property243 def spool_bytes(self):244 """读取仅本索引新增的累计配额。参数: 无。返回: int。异常: 无，不含BLF或发布副本。"""245 return self._spool_bytes246247 def _require(self, state):248 """检查活状态，半关闭或失败前缀不能继续使用。249250 参数: state为building/sealed。返回: None。异常: ResultClosedError/RuntimeError。251 """252 if self._closed or self._closing:253 raise ResultClosedError('原始导出仓已关闭或正在清理')254 if self.state != state:255 raise RuntimeError('原始导出仓状态不允许操作: ' + self.state)256257 def _fail(self, exc):258 """保原主错/取消、failed状态和最高恢复owner，清理副错只记note。259260 参数: exc为实际主异常。返回: None。异常: 清理错误不替代原exc。261 """262 self.state = 'failed'263 exc.raw_export_store_cleanup_owner = self264 _cleanup_owner(self, exc, 'raw_export_store_cleanup_owner', '原始导出仓')265266 def next_path(self):267 """计算当前记录对应BLF路径，不创建文件、不跳过无观测索引。268269 参数: 无。返回: 根/raw_segments/{record_count:012d}.blf。270 异常: 状态错误或索引超过12位能力；原writer由调用方独立持有和关闭。271 """272 self._require('building')273 _integer(self.record_count, 'raw_exports.record_count', _MAX_INDEX)274 return self._raw_path / f'{self.record_count:012d}.blf'275276 def _relation(self, record, index):277 """核本仓归属和连续索引，不从文件名猜来源。278279 参数: record为冻结记录；index为当前预期号。返回: None。异常: TypeError/ValueError。280 """281 if type(record) is not RawExportRecord:282 raise TypeError('原始导出仓仅接受RawExportRecord')283 if (record.index != index or record.source != self.source or record.session_id != self.session_id284 or record.device_label != self.device_label):285 raise ValueError('原始导出记录索引/来源/会话/设备不属于本仓')286287 def _close_fd(self, slot):288 """真实FD关闭后才撤槽，关闭后抛错也依据actual.closed处理。289290 参数: slot为本owner固定FD槽。返回: None。异常: 原close/取消，持续失败可重试。291 """292 handle = getattr(self, slot)293 if handle is not None:294 try:295 handle.close()296 finally:297 if handle.closed:298 setattr(self, slot, None)299300 def _file(self, record, destination=None):301 """有限块核实际BLF；可同时复制到独占目标，不调用任何CAN读写器。302303 参数: record为exported记录；destination为None或新文件Path。返回: None。304 异常: 原预算/IO/取消或ResultFormatError；所有当前FD仍从本store可达。305 """306 if self._input_fd is not None or self._output_fd is not None or self._io_cost:307 raise ResultClosedError('原始导出仓有未完成文件资源待清理')308 _reserve_owned_ram(self, _IO_COST, '_io_cost')309 try:310 self._input_fd = (self.path / record.file_name).open('rb', buffering=0)311 except FileNotFoundError as exc:312 # 索引已指明存在的原件消失是输入失效；不能当成本次发布目录错误重试。313 self._fail(exc)314 raise315 if destination is not None:316 self._output_fd = destination.open('xb', buffering=0)317 digest, count = sha256(), 0318 # 原件和副本同时只持一个块；超过声明字节立即拒绝，不扫无界额外后缀。319 while True:320 self._chunk = self._input_fd.read(_CHUNK_BYTES)321 if not self._chunk:322 break323 count += len(self._chunk)324 if count > record.byte_count:325 exc = ResultFormatError('原始BLF字节数超过索引')326 self._fail(exc)327 raise exc328 digest.update(self._chunk)329 if self._output_fd is not None:330 _write_all(self._output_fd, self._chunk)331 if count != record.byte_count or digest.hexdigest() != record.sha256.lower():332 exc = ResultFormatError('原始BLF字节数或SHA与索引不符')333 self._fail(exc)334 raise exc335 if self._output_fd is not None:336 self._output_fd.flush()337 self._close_fd('_output_fd')338 self._close_fd('_input_fd')339 # 发布字节也以真实文件回读核验，不能只凭write返回长度宣称副本正确。340 if destination is not None:341 self._check_copy(destination, record.byte_count, record.sha256)342 self._chunk = None343 _release_owned_ram(self, self._io_cost, '_io_cost')344345 def _check_copy(self, path, expected_bytes, expected_sha):346 """在已预约IO空间内回读真实发布文件，BLF与索引使用同一有界验证。347348 参数: path为本次新目标Path；expected_bytes/SHA为已封口原事实。返回: None。349 异常: 原打开/read/close/取消或ResultFormatError；当前真实FD留在本owner槽。350 """351 if self._input_fd is not None or not self._io_cost:352 raise ResultClosedError('发布文件回读需空输入槽和已预约IO空间')353 self._input_fd = path.open('rb', buffering=0)354 digest, count = sha256(), 0355 while True:356 self._chunk = self._input_fd.read(_CHUNK_BYTES)357 if not self._chunk:358 break359 count += len(self._chunk)360 if count > expected_bytes:361 raise ResultFormatError('发布文件字节数超过封口事实')362 digest.update(self._chunk)363 if count != expected_bytes or digest.hexdigest() != expected_sha.lower():364 raise ResultFormatError('发布文件字节数或SHA与封口事实不符')365 self._close_fd('_input_fd')366 self._chunk = None367368 def append(self, record):369 """预付索引RAM/累计spool，核实际文件后完整写一行再推进计数。370371 参数: record为本仓当前索引的RawExportRecord。返回: None。372 异常: 原归属/预算/文件/编码/取消；开始后的失败poison，原BLF费用不重复预约。373 """374 self._require('building')375 try:376 self._relation(record, self.record_count)377 _reserve_owned_ram(self, _ROW_COST, '_append_cost')378 self._payload = encode_packet(_packet(record)) + b'\n'379 before = self._budget.spool_bytes380 try:381 self._budget.reserve_spool(len(self._payload))382 finally:383 if self._budget.spool_bytes == before + len(self._payload):384 self._spool_bytes += len(self._payload)385 if record.status == 'exported':386 self._file(record)387 _write_all(self._writer, self._payload)388 self._digest.update(self._payload)389 self.inventory_bytes += len(self._payload)390 self.record_count += 1391 self.file_count += int(record.status == 'exported')392 self._payload = None393 _release_owned_ram(self, self._append_cost, '_append_cost')394 except BaseException as exc:395 self._fail(exc)396 raise397398 def seal(self):399 """真实关闭索引writer并固定流SHA；零记录亦可封口。400401 参数: 无。返回: None。异常: 原flush/close/取消，失败不能冒充sealed。402 """403 self._require('building')404 try:405 self._writer.flush()406 self._close_fd('_writer')407 self.sha256, self.state = self._digest.hexdigest(), 'sealed'408 except BaseException as exc:409 self._fail(exc)410 raise411412 def iter_records(self):413 """创建有限独立索引reader，未返回构造始终在本父pending槽可达。414415 参数: 无。返回: 逐条RawExportRecord迭代器。异常: 原预算/打开/登记/取消及状态错误。416 """417 self._require('sealed')418 if self._pending_reader is not None:419 raise ResultClosedError('原始导出仓有未返回reader待清理')420 try:421 reader = _RawExportReader(self)422 self._pending_reader = None423 return reader424 except BaseException as exc:425 self._fail(exc)426 raise427428 def copy_to(self, destination):429 """向新发布根独占创建原始目录与索引，逐条核验复制现成BLF。430431 参数: destination为Path；根可已有普通结果表，raw_segments及索引必须均不存在。432 返回: 目标raw_segments.jsonl Path。异常: 原路径/预算/IO/取消或完整性错误；433 输出失败保本次FD责任及部分文件，不覆盖已有工件或关闭sealed输入；434 入口先恢复上次复制，持续关闭失败时不得覆盖旧槽或创建下个目标。435 """436 self._require('sealed')437 self.recover_copy()438 if not isinstance(destination, Path):439 raise TypeError('原始导出发布根需要Path')440 if sum(len(part) for part in destination.parts) > MAX_WIRE_BYTES:441 raise ValueError('原始导出发布路径超过有界能力')442 target = destination.resolve()443 raw_path, inventory = target / 'raw_segments', target / 'raw_segments.jsonl'444 if raw_path.exists() or inventory.exists():445 raise FileExistsError('原始导出发布目录或索引已存在')446 try:447 _reserve_owned_ram(self, _ROW_COST + 32 * len(str(target)), '_copy_cost')448 target.mkdir(parents=True, exist_ok=True)449 raw_path.mkdir(exist_ok=False)450 self._inventory_fd = inventory.open('xb', buffering=0)451 # 本次临时reader先装独立copy槽；其预约/关闭取消不破坏其他已封口输入。452 self._copy_reader = _RawExportReader(self, copy=True)453 digest, count, files = sha256(), 0, 0454 # reader只交付当前一行；不累计文件路径或导出记录列表。455 for record in self._copy_reader:456 if record.status == 'exported':457 self._file(record, target / record.file_name)458 files += 1459 self._payload = encode_packet(_packet(record)) + b'\n'460 _write_all(self._inventory_fd, self._payload)461 digest.update(self._payload)462 self._payload = None463 count += 1464 if (not self._copy_reader.verified_eof or count != self.record_count465 or files != self.file_count or digest.hexdigest() != self.sha256):466 exc = ResultFormatError('原始导出复制索引未完整匹配封口事实')467 self._fail(exc)468 raise exc469 self._copy_reader = None470 self._inventory_fd.flush()471 self._close_fd('_inventory_fd')472 _reserve_owned_ram(self, _IO_COST, '_io_cost')473 self._check_copy(inventory, self.inventory_bytes, self.sha256)474 _release_owned_ram(self, self._io_cost, '_io_cost')475 _release_owned_ram(self, self._copy_cost, '_copy_cost')476 return inventory477 except BaseException as exc:478 # 输入reader或原BLF摘要错误可已poison父；输出故障只收回本次复制责任。479 exc.raw_export_store_cleanup_owner = self480 for attempt in range(2):481 try:482 self._close_copy_resources()483 break484 except BaseException as cleanup:485 exc.add_note(f'原始导出复制清理第{attempt + 1}次失败: {type(cleanup).__name__}')486 raise487488 def recover_copy(self):489 """仅恢复上次输出复制的真实责任，保sealed固定输入及其他独立reader。490491 参数: 无。返回: None；copy_to入口会自动调用，调用方也可显式恢复。492 异常: 原FD/reader/费用关闭错误或状态错误；持续失败保最高store与原槽，493 不把整个输入仓置关闭中，不覆盖上次复制尚未释放的资源。494 """495 self._require('sealed')496 self._close_copy_resources()497498 def _close_copy_resources(self):499 """关闭本次copy子reader和FD，完全释放后再清当前缓冲与复制费用。500501 参数: 无。返回: None；不操作固定费、原索引writer或其他独立reader。502 异常: 首个真实关闭/抵销错误，副错记note；最高store保存未完成实际槽。503 """504 failure = None505 if self._copy_reader is not None:506 try:507 self._copy_reader.close()508 except BaseException as exc:509 failure = exc510 finally:511 if self._copy_reader.closed:512 self._copy_reader = None513 for slot in ('_input_fd', '_output_fd', '_inventory_fd'):514 try:515 self._close_fd(slot)516 except BaseException as exc:517 if failure is None:518 failure = exc519 else:520 failure.add_note('原始导出本次复制FD关闭附加失败: ' + type(exc).__name__)521 if self._copy_reader is None and all(getattr(self, slot) is None for slot in ('_input_fd', '_output_fd', '_inventory_fd')):522 self._payload = self._chunk = None523 for slot in ('_io_cost', '_copy_cost'):524 try:525 _release_owned_ram(self, getattr(self, slot), slot)526 except BaseException as exc:527 if failure is None:528 failure = exc529 else:530 failure.add_note('原始导出本次复制费用抵销附加失败: ' + type(exc).__name__)531 if failure is not None:532 failure.raw_export_store_cleanup_owner = self533 raise failure534535 def close(self):536 """关闭全部自有FD/reader并抵销自有RAM，文件与累计spool保留。537538 参数: 无。返回: None，完成后幂等。异常: 首个真实关闭/抵销错误，副错记note；539 持续故障保原槽/最高owner，不能从借用账本总额倒推或多扣费用。540 """541 if self._closed:542 return543 self._closing = True544 if self.state == 'building':545 self.state = 'failed'546 failure = None547 # pending与活登记可短暂指向同一真实子；先pending，成功后从登记中解除。548 readers = []549 for reader in (self._pending_reader, self._copy_reader, *tuple(self._readers or ())):550 if reader is not None and reader not in readers:551 readers.append(reader)552 for reader in readers:553 try:554 reader.close()555 except BaseException as exc:556 if failure is None:557 failure = exc558 else:559 failure.add_note('原始导出reader关闭附加失败: ' + type(exc).__name__)560 if self._copy_reader is not None and self._copy_reader.closed:561 self._copy_reader = None562 for slot in ('_writer', '_input_fd', '_output_fd', '_inventory_fd'):563 try:564 self._close_fd(slot)565 except BaseException as exc:566 if failure is None:567 failure = exc568 else:569 failure.add_note('原始导出FD关闭附加失败: ' + type(exc).__name__)570 if (self._pending_reader is None and not self._readers and self._copy_reader is None571 and all(getattr(self, slot) is None for slot in ('_writer', '_input_fd', '_output_fd', '_inventory_fd'))):572 self._payload = self._chunk = None573 for slot in ('_append_cost', '_io_cost', '_copy_cost', '_fixed_cost'):574 try:575 _release_owned_ram(self, getattr(self, slot), slot)576 except BaseException as exc:577 if failure is None:578 failure = exc579 else:580 failure.add_note('原始导出费用抵销附加失败: ' + type(exc).__name__)581 if self._held_cost == 0:582 self._closed = True583 if failure is not None:584 failure.raw_export_store_cleanup_owner = self585 raise failure586587588class _RawExportReader:589 """拥有单个索引位置/摘要和固定读槽；完整EOF与成功关闭才verified_eof。590591 copy模式只把临时资源错误交本次复制恢复，输入格式/摘要破坏仍poison父仓。592 """593594 def __init__(self, owner, *, copy=False):595 """先装父pending槽，再付费打开/登记真实reader。596597 参数: owner为活sealed仓；copy为内部临时复制模式，直接先装父_copy_reader槽。598 返回: 独立reader。异常: 原预算/打开/登记/取消；真实父槽始终保未返回self。599 """600 self.owner, self._budget = owner, owner.budget601 self._copy = copy602 self.handle = self._digest = None603 self._fixed_cost = self._held_cost = self._count = self._files = self._bytes = 0604 self.closed = self._closing = self.verified_eof = False605 if copy:606 owner._copy_reader = self607 else:608 owner._pending_reader = self609 # 仅索引reader允许8192字节有限预读；buffer/对象另付16384，原行限及真EOF不变。610 # BLF分块复制与写端不改，同一handle始终拥有buffer及真实raw FD。611 _reserve_owned_ram(self, _ROW_COST + 16384, '_fixed_cost')612 self._digest = sha256()613 try:614 self.handle = owner.inventory_path.open('rb', buffering=8192)615 except FileNotFoundError as exc:616 # 普通工厂原已有_fail；copy专用工厂也须把源索引消失归为输入完整性失效。617 if copy:618 owner._fail(exc)619 raise620 owner._readers.add(self)621622 def __iter__(self):623 """保持独立原位置。参数: 无。返回: self。异常: 无，不重新开文件。"""624 return self625626 def __next__(self):627 """逐行校索引归属/序号，真实EOF核完整SHA、条数、文件数与字节数后关闭。628629 参数: 无。返回: RawExportRecord。异常: 真实StopIteration或原格式/关闭/取消；630 输入损坏poison父；copy临时资源失败交本次恢复，原异常/取消对象保持。631 """632 if self.closed:633 raise StopIteration634 self.owner._require('sealed')635 if self._closing:636 raise ResultClosedError('原始导出reader正在清理')637 try:638 payload = self.handle.readline(MAX_WIRE_BYTES + 1)639 if payload:640 if len(payload) > MAX_WIRE_BYTES or not payload.endswith(b'\n'):641 raise ResultFormatError('原始导出索引行超限或缺完整换行')642 if self._count >= self.owner.record_count:643 raise ResultFormatError('原始导出索引出现额外后缀')644 record = _decode(payload[:-1])645 try:646 self.owner._relation(record, self._count)647 except (TypeError, ValueError) as exc:648 raise ResultFormatError('原始导出索引归属或序号非法') from exc649 self._digest.update(payload)650 self._count += 1651 self._files += int(record.status == 'exported')652 self._bytes += len(payload)653 return record654 if (self._count != self.owner.record_count or self._files != self.owner.file_count655 or self._bytes != self.owner.inventory_bytes or self._digest.hexdigest() != self.owner.sha256):656 raise ResultFormatError('原始导出索引EOF数量或SHA不匹配')657 self.close()658 except BaseException as exc:659 if not self._copy or isinstance(exc, ResultFormatError):660 self.owner._fail(exc)661 else:662 exc.raw_export_store_cleanup_owner = self.owner663 raise664 self.verified_eof = True665 raise StopIteration666667 def close(self):668 """真实关FD、抵销本笔费用并解除父登记后才closed，可持续重试。669670 参数: 无。返回: None。异常: 原close/预算/取消；最高store仍持本实例。671 """672 if self.closed:673 return674 self._closing = True675 try:676 if self.handle is not None and not self.handle.closed:677 self.handle.close()678 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')679 self.owner._readers.discard(self)680 if self.owner._pending_reader is self:681 self.owner._pending_reader = None682 self.closed = True683 except BaseException as exc:684 exc.raw_export_store_cleanup_owner = self.owner685 raise686