chronaxis.legacy.message_store
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""独立封存一次解码返回的完整原值，供后续消息分析回读。23append输入LegacyDecodedFrame、已显式映射的int64纳秒及量化标志；磁盘每行4保留固定store_version/time_ns/time_quantized外壳和原decoded包。内层仍用5现有64KiB严格records编码，外层只额外允许1024字节且包含换行，不改变IPC。6所有原帧、通道类型、原字段、0/False及非有限诊断均保留；不筛选、去重、7转换时间或数值，不生成evaluation_index或执行计算；求值记录由消息分析层产生。89完整写入才推进条数和流SHA；任意追加错误poison，封口须writer实际关闭。10独立reader按原序读取，真正EOF核全SHA/count并成功关闭FD及费用后才给11verified_eof。sealed仅证明本仓完整封口，不证明原来源已耗尽。1213固定状态、单行编解码与各活reader均在分配前预付；只保存当前行和活reader，14返回记录和返回异常中的原wire有界诊断由调用方拥有并承担保留空间，预算估算15不含这些输出、调用方已有输入或整个进程RSS。close只关16本owner真实FD和实际费用，文件及累计磁盘预留保留，不抵销借用账本其他费用。17本模块只依赖标准库、现有冻结模型与轻量budget，不导入NumPy或旧CAN库。18若work_dir已是调用方Path，其首次str表示可能缓存于该借用输入；这部分表示与19原Path归调用方。本仓新建Path、语法验证及resolve等均在自身预付之后执行。20显式sparse_sequences只保存调用方选定的原记录，sequence保原值且严格递增；21count/SHA/line_limit均描述实际保存内容，不证明全输入条数或未保存内容。22"""2324# 原包模型与严格JSON边界复用现有实现，文件与预算责任只由本仓承担。25from dataclasses import dataclass26from hashlib import sha25627from pathlib import Path2829from ..core.validation import require_identifier, require_text30from ..results.budget import ResourceBudget, _cleanup_owner, _release_owned_ram, _reserve_owned_ram31from ..results.errors import ResultClosedError, ResultFormatError32from ..timebase.conversion import require_int64_ns33from .records import MAX_WIRE_BYTES, LegacyDecodedFrame, _decoded_from_packet, decoded_to_dict34from .records import decode_packet, encode_packet353637# 外壳大小独立于原IPC；含换行的磁盘行始终不能超过此明确上限。38MAX_DECODED_MESSAGE_ROW_BYTES = MAX_WIRE_BYTES + 102439_HEADER_LIMIT = 102440_DECODED_BOUNDARY = b',"decoded":'414243def _clear_records_traceback(exc: BaseException) -> None:44 """断开当前错误中已结束records函数的临时引用，保原错误及traceback结构。4546 参数: exc为当前编码或解析原异常，只沿其自身__traceback__逐节点检查。47 返回: None；只清精确chronaxis.legacy.records模块的已结束frame局部引用。48 异常: 活跃frame.clear的RuntimeError表示不可清，原样跳过；不递归cause/context，49 不清用户、测试、stdlib、CAN或本仓活跃帧，也不删JSON/Unicode错误诊断字段。50 """51 # 不收集TB或异常链，避免因错误context环或历史异常建立无界遍历/visited状态。52 current = exc.__traceback__53 while current is not None:54 frame = current.tb_frame55 if frame.f_globals.get('__name__') == 'chronaxis.legacy.records':56 try:57 frame.clear()58 except RuntimeError:59 pass60 current = current.tb_next616263@dataclass(frozen=True)64class StoredDecodedMessage:65 """冻结一条已映射时间的完整解码返回，不制造旧或新求值序号。6667 time_ns仅按共享时间协议规范，time_quantized必须由调用方明确提供。68 decoded保原LegacyDecodedFrame对象；合法值域沿原records，不额外开放69 None/str解码值。原channel的None/str/int类型不受目标帧通道规则改写。70 """7172 time_ns: int # 调用方显式映射后的会话坐标，不在仓中再次加偏移。73 time_quantized: bool # 原转换的量化事实，不由浮点时间猜测。74 decoded: LegacyDecodedFrame # 一次原yield和parse的完整冻结返回。7576 def __post_init__(self) -> None:77 """核对单条存储模型，保原decoded引用及独立端点范围。7879 参数: 无，读取三个构造字段。80 返回: None；只将time_ns规范成原生int，不重建decoded或计算时间跨度。81 异常: TypeError为类型不符；TimeOverflowError为时间超int64。82 """83 # 模型不要求通道为int；那是后续目标帧政策，不能在原序仓提前施加。84 mapped = require_int64_ns(self.time_ns)85 if type(self.time_quantized) is not bool:86 raise TypeError('time_quantized必须为原生bool')87 if type(self.decoded) is not LegacyDecodedFrame:88 raise TypeError('decoded必须为LegacyDecodedFrame')89 object.__setattr__(self, 'time_ns', mapped)909192def _encode_row(record: LegacyDecodedFrame, time_ns: int, time_quantized: bool) -> bytes:93 """分别严格编码小外壳和原decoded包，再拼固定的decoded末字段。9495 参数: record为原冻结返回；time_ns和time_quantized已经由append校验。96 返回: bytes；完整JSON对象加换行，原decoded payload不超过MAX_WIRE_BYTES。97 异常: 原records类型、限额、编码、分配或取消异常，不修改全局wire限额。98 """99 header = packet = inner = result = None100 try:101 # header无自由文本或嵌套；原值树始终单独经过原64KiB边界。102 header = encode_packet({'store_version': 1, 'time_ns': time_ns,103 'time_quantized': time_quantized})104 packet = decoded_to_dict(record)105 inner = encode_packet(packet)106 result = header[:-1] + _DECODED_BOUNDARY + inner + b'}\n'107 return result108 except BaseException as exc:109 # 原codec失败frame也可能保留payload树；只清其已结束局部，不替换异常或TB节点。110 _clear_records_traceback(exc)111 raise112 finally:113 # 调用方可能长期保留原异常，不能由本层frame继续保活编码副本。114 header = packet = inner = result = None115116117def _decode_row(payload: bytes) -> StoredDecodedMessage:118 """在明确磁盘上限内拆固定边界，分别调用原严格包解析器。119120 参数: payload为带完整换行的有界磁盘行，decoded必须为最后字段。121 返回: StoredDecodedMessage，原diagnostic标签按现records模型恢复。122 异常: ResultFormatError为外壳、重复/多余字段、原包或模型不合法；123 内存或取消错误保持原样，不用不受限JSON解析绕过IPC边界。124 """125 header_bytes = inner_bytes = header = decoded = None126 try:127 # 外层只允许固定小header；取第一个边界，不被原信号名中的同形文本干扰。128 if not payload.endswith(b'}\n'):129 raise ResultFormatError('解码仓行缺少完整外层或换行')130 boundary = payload.find(_DECODED_BOUNDARY)131 if boundary < 0 or boundary + 1 > _HEADER_LIMIT:132 raise ResultFormatError('解码仓固定decoded边界缺失或header超限')133 header_bytes = payload[:boundary] + b'}'134 inner_bytes = payload[boundary + len(_DECODED_BOUNDARY):-2]135 header = decode_packet(header_bytes)136 if set(header) != {'store_version', 'time_ns', 'time_quantized'}:137 raise ResultFormatError('解码仓外壳字段缺失、多余或decoded位置错误')138 if type(header['store_version']) is not int or header['store_version'] != 1:139 raise ResultFormatError('解码仓store_version必须为原生整数1')140 # 小header先校验完，坏端点或量化标志不应先生成一份大decoded记录。141 mapped = require_int64_ns(header['time_ns'], 'message_store.time_ns')142 if type(header['time_quantized']) is not bool:143 raise ResultFormatError('解码仓time_quantized必须为原生bool')144 # 原payload仍独立核完整JSON/模型及64KiB/ITEM上限；固定小包可用正证明145 # 省重复遍历，复杂输入回原严格入口，不将整个超64KiB磁盘行当作原包解析。146 decoded = _decoded_from_packet(inner_bytes)147 return StoredDecodedMessage(mapped, header['time_quantized'], decoded)148 except (TypeError, ValueError, OverflowError) as exc:149 # 结构错误保为格式错误的原cause；先结束records帧的临时责任，再由父收尾费用。150 _clear_records_traceback(exc)151 raise ResultFormatError('解码仓行结构或原decoded包损坏') from exc152 except BaseException as exc:153 # 取消/内存错误不包装，也只处理本次TB中的records临时引用。154 _clear_records_traceback(exc)155 raise156 finally:157 # 返回对象由调用方拥有；失败时本层临时字典和bytes不留在traceback中。158 payload = header_bytes = inner_bytes = header = decoded = None159160161class DecodedMessageStore:162 """拥有一个新JSONL、封口事实和所有独立reader的实际资源责任。163164 state为building/sealed/failed/cancelled；closed仅在自有FD和费用全部完成后真。165 _pending_reader在子构造副作用前安装，_readers只保活reader。固定/append槽166 与累计_spool_bytes各自记真实交易，不由共享账本余额猜测本owner归属。167 """168169 def __init__(self, budget, work_dir, source_name, session_id, *, decoder_profile,170 read_identity, max_row_bytes=MAX_DECODED_MESSAGE_ROW_BYTES, sparse_sequences=False, integrity=True):171 """预付有限状态并在显式全新目录打开唯一writer，拒绝已有目录。172173 参数: budget为借用ResourceBudget；work_dir为新原生str/Path；source_name/session_id174 为原生str简单名称；decoder_profile/read_identity为原生非空文本；max_row_bytes为175 1至MAX_DECODED_MESSAGE_ROW_BYTES，包含换行；sparse_sequences显式允许原序号有间隔，176 仍严格递增，count/SHA仅表示实际保存的记录，默认仍要求从0连续。177 返回: building仓；不读取BLF/DBC或证明来源EOF。178 异常: 类型、路径、预算、文件或取消原错；持续清理失败保最高179 decoded_message_store_cleanup_owner，恢复后对同一owner.close。180 """181 if not isinstance(budget, ResourceBudget) or not (type(work_dir) is str or isinstance(work_dir, Path)):182 raise TypeError('解码仓需要ResourceBudget及显式新目录')183 # 先装全部真实责任槽，之后reserve更新后取消或mkdir失败都可从本owner恢复。184 if type(integrity) is not bool: raise TypeError('integrity需要bool')185 self.integrity=integrity186 self.budget = self._budget = budget187 self.source_name, self.session_id = source_name, session_id188 self.decoder_profile, self.read_identity = decoder_profile, read_identity189 self.max_row_bytes = max_row_bytes190 self.sparse_sequences, self._last_sequence = sparse_sequences, -1191 self._line_limit = 1192 self.path = self.work_dir = self.file_path = None193 self._writer = self._pending_reader = self._pending_payload = None194 self._readers = self._digest = None195 self._fixed_cost = self._append_cost = self._held_cost = self._spool_bytes = 0196 self.count, self.sha256, self._last_time_ns = 0, None, None197 self.state, self._closed, self._closing = 'building', False, False198 try:199 # 先用无副本类型及长度估费，避免长文本strip或新Path解析先于预算拒绝。200 if any(type(value) is not str for value in (source_name, session_id, decoder_profile, read_identity)):201 raise TypeError('解码仓来源、会话、profile和read_identity需要原生str')202 if type(sparse_sequences) is not bool:203 raise TypeError('sparse_sequences必须为原生bool')204 if type(max_row_bytes) is not int or not 1 <= max_row_bytes <= MAX_DECODED_MESSAGE_ROW_BYTES:205 raise ValueError('解码仓行上限需要合法范围内的原生int')206 # 原生str直接取len；已有Path的str缓存属于借用输入，不创建本仓Path副本。207 path_length = len(work_dir) if type(work_dir) is str else len(str(work_dir))208 cost = 65536 + 32 * (path_length + len(source_name) + len(session_id)209 + len(decoder_profile) + len(read_identity))210 _reserve_owned_ram(self, cost, '_fixed_cost')211 # 费用已实际归属后才允许语法验证的strip临时文本和本仓路径构造。212 require_identifier(source_name, 'message_store.source_name')213 require_identifier(session_id, 'message_store.session_id')214 require_text(decoder_profile, 'decoder_profile')215 require_text(read_identity, 'read_identity')216 requested = Path(work_dir)217 # 先解析实际路径并检查存在性，不写已有目录或错误诊断文件。218 resolved = requested.resolve()219 if resolved.exists():220 raise FileExistsError(resolved)221 self.path = self.work_dir = resolved222 self._readers, self._digest = set(), sha256() if integrity else None223 resolved.mkdir(parents=True, exist_ok=False)224 self.file_path = resolved / 'decoded_messages.jsonl'225 self._writer = self.file_path.open('xb', buffering=0)226 except BaseException as exc:227 self._fail(exc)228 raise229230 @property231 def line_limit(self):232 """读取完整成功写入行的实际最大字节数，空仓取1以保持有限读上界。233234 参数: 无。返回: 正int，只由本仓成功append推进；公开硬限仍为max_row_bytes。235 异常: 无；文件损坏、未完成写入和外部stat都不会提高此值。236 """237 return self._line_limit238239 @property240 def closed(self):241 """读取实际完整关闭事实；参数: 无；返回: bool；异常: 无。"""242 return self._closed243244 @property245 def spool_bytes(self):246 """读取本owner累计磁盘预留；参数: 无；返回: bytes整数；异常: 无。"""247 return self._spool_bytes248249 def _require(self, state):250 """校验活owner及所需构建状态，不把关闭或poison伪装成空流。251252 参数: state为building或sealed。返回: None。253 异常: ResultClosedError为关闭中；RuntimeError为状态不匹配。254 """255 if self._closed or self._closing:256 raise ResultClosedError('解码仓已关闭或正在清理')257 if self.state != state:258 raise RuntimeError('解码仓状态不允许此操作: ' + self.state)259260 def _fail(self, exc):261 """保原异常并poison，有限尝试清理，持续失败保最高真实仓owner。262263 参数: exc为主异常。返回: None。264 异常: 清理副错只加notes，不覆盖同一普通错误、KeyboardInterrupt或SystemExit。265 """266 self.state = 'cancelled' if isinstance(exc, (KeyboardInterrupt, SystemExit)) else 'failed'267 exc.decoded_message_store_cleanup_owner = self268 _cleanup_owner(self, exc, 'decoded_message_store_cleanup_owner', '解码仓')269270 def _reserve_spool(self, amount):271 """写入前累计本笔磁盘预留，更新后取消也记录实际归属且不倒扣文件。272273 参数: amount为本行bytes。返回: None。274 异常: 原配额或取消错误；未入账不增加本owner累计。275 """276 before = self._budget.spool_bytes277 try:278 self._budget.reserve_spool(amount)279 finally:280 if self._budget.spool_bytes == before + amount:281 self._spool_bytes += amount282283 def _write_pending(self):284 """处理真实短写，并在错误传播前释放所有新建memoryview引用。285286 参数: 无；读取本owner已预付、已预约磁盘的_pending_payload。287 返回: None，全部字节已交真实writer。288 异常: 原写入/取消错误或OSError无推进；失败前缀由父保留并poison。289 """290 view = part = None291 try:292 # 使用可释放view，避免短写切片复制及错误wrapper保活本层待写bytes。293 view = memoryview(self._pending_payload)294 offset = 0295 while offset < len(view):296 try:297 part = view[offset:]298 written = self._writer.write(part)299 if type(written) is not int or not 0 < written <= len(part):300 raise OSError('解码仓写入未推进或返回非法长度')301 offset += written302 finally:303 if part is not None:304 part.release()305 part = None306 finally:307 if view is not None:308 view.release()309 view = part = None310311 def append(self, record, time_ns, *, time_quantized):312 """完整保存一次原yield，原序和完整写入成立后才推进count/SHA。313314 参数: record为LegacyDecodedFrame；time_ns已由调用方映射；time_quantized315 为明确原生bool。全部channel和值沿原模型，不套目标帧或求值规则。316 返回: None。异常: 类型、时间、顺序、编码、行长、预算、磁盘、写入或取消317 原错；任意追加失败poison，最高owner可恢复但不能再seal或读前缀。318 """319 self._require('building')320 try:321 # 身份序号必须0起连续，时间可同刻；只校单端点，不计算可能溢出的跨度。322 if type(record) is not LegacyDecodedFrame:323 raise TypeError('解码仓仅接受LegacyDecodedFrame')324 mapped = require_int64_ns(time_ns, 'message_store.time_ns')325 if type(time_quantized) is not bool:326 raise TypeError('time_quantized必须为原生bool')327 if (record.sequence <= self._last_sequence if self.sparse_sequences else record.sequence != self.count) or self._last_time_ns is not None and mapped < self._last_time_ns:328 raise ValueError('解码仓sequence不满足声明连续/稀疏顺序或time_ns回退')329 # 按完整能力预付编码拒绝路径，不因用户设置很小的行上限漏算内层包工作区。330 _reserve_owned_ram(self, 65536 + 64 * MAX_DECODED_MESSAGE_ROW_BYTES, '_append_cost')331 self._pending_payload = _encode_row(record, mapped, time_quantized)332 if len(self._pending_payload) > self.max_row_bytes:333 raise ValueError('解码仓行超过max_row_bytes')334 self._reserve_spool(len(self._pending_payload))335 self._write_pending()336 # 当前行已完整写入；后续任何故障仍poison，不把部分或不确定状态重新开放。337 if self._digest is not None:338 self._digest.update(self._pending_payload)339 self.count += 1340 self._line_limit = max(self._line_limit, len(self._pending_payload))341 self._last_time_ns, self._last_sequence = mapped, record.sequence342 self._pending_payload = None343 _release_owned_ram(self, self._append_cost, '_append_cost')344 except BaseException as exc:345 self._pending_payload = None346 self._fail(exc)347 raise348349 def _close_writer(self):350 """按真实FD关闭事实清writer槽，持续close前失败保留原handle。351352 参数: 无。返回: None。353 异常: 原FD或取消错误；实际closed后抛错也不重关已经完成的handle。354 """355 # close返回或抛错都检查同一真实handle；只在实际closed后撤除父writer槽。356 if self._writer is not None:357 try:358 self._writer.close()359 finally:360 if self._writer.closed:361 self._writer = None362363 def seal(self):364 """flush并实际关闭writer后才固定SHA及sealed，不证明原来源EOF。365366 参数: 无。返回: None；空仓也可完整封口。367 异常: 原flush、close、摘要或取消错误；poison后不再读或重试seal。368 """369 self._require('building')370 try:371 # 先完成真实输入FD责任，再开放可读事实，不能只因最后append成功就sealed。372 self._writer.flush()373 self._close_writer()374 self.sha256, self.state = (self._digest.hexdigest() if self._digest is not None else None), 'sealed'375 except BaseException as exc:376 self._fail(exc)377 raise378379 def iter_records(self):380 """创建独立有预算原序reader，构造未返回责任先由本父真实槽保存。381382 参数: 无。返回: 逐条StoredDecodedMessage reader；各自位置互不干扰。383 异常: 状态、预算、打开、登记或取消原错，最高仓owner可恢复全部实际子资源。384 """385 self._require('sealed')386 try:387 reader = _DecodedMessageReader(self)388 self._pending_reader = None389 return reader390 except BaseException as exc:391 self._fail(exc)392 raise393394 def close(self):395 """可重试关闭所有自有stream及实际费用，保留文件与累计磁盘申请。396397 参数: 无。返回: None；所有责任闭合才closed=True，重复close幂等。398 异常: 首个关闭/抵销错误原样传播，其他错误只note；最高owner不变，不关闭399 借用budget或释放其他owner费用，不删除文件及已交付调用方的记录。400 """401 # 完全关闭才幂等返回；开始关闭立即拒新操作，未封口构建只留下取消前缀。402 if self._closed:403 return404 self._closing = True405 if self.state == 'building':406 self.state = 'cancelled'407 first_error = None408 # 未返回实例先关；成功时子会解除登记，剩余活stream再逐一处理。409 if self._pending_reader is not None:410 try:411 self._pending_reader.close()412 except BaseException as exc:413 first_error = exc414 readers = None415 try:416 # 当前活stream数受各reader费用限制；快照分配失败也保留父实际登记继续恢复。417 readers = () if self._readers is None else tuple(self._readers)418 for reader in readers:419 try:420 reader.close()421 except BaseException as exc:422 if first_error is None:423 first_error = exc424 else:425 first_error.add_note('解码仓其他reader关闭附加失败: ' + type(exc).__name__)426 except BaseException as exc:427 if first_error is None:428 first_error = exc429 else:430 first_error.add_note('解码仓reader关闭快照附加失败: ' + type(exc).__name__)431 finally:432 # 原异常不保活一个已抵费的全stream快照，真正未关实例仍在父登记中。433 readers = None434 try:435 self._close_writer()436 except BaseException as exc:437 if first_error is None:438 first_error = exc439 else:440 first_error.add_note('解码仓writer关闭附加失败: ' + type(exc).__name__)441 # 每个活reader费用覆盖其登记及关闭快照；父固定费必须等所有子和writer闭合。442 if self._writer is None and self._pending_reader is None and not self._readers:443 self._pending_payload = None444 for slot in ('_append_cost', '_fixed_cost'):445 try:446 _release_owned_ram(self, getattr(self, slot), slot)447 except BaseException as exc:448 if first_error is None:449 first_error = exc450 else:451 first_error.add_note('解码仓费用抵销附加失败: ' + type(exc).__name__)452 if self._held_cost == 0:453 self._closed = True454 if first_error is not None:455 first_error.decoded_message_store_cleanup_owner = self456 raise first_error457458459class _DecodedMessageReader:460 """拥有独立FD、SHA/count和有界单行空间，不持有已交付记录列表。461462 verified_eof只在真实EOF、完整SHA/count和实际FD/费用/登记全部关闭后为True。463 提前close不产生证明；关闭中拒绝读，未返回构造实例先由父pending槽接管。464 """465466 def __init__(self, owner):467 """先安装真实未返回子槽，再预约、打开和登记独立reader。468469 参数: owner为活sealed DecodedMessageStore。470 返回: None；初始化实例状态；父工厂成功返回前一直可从_pending_reader找到本实例。471 异常: 原预算、打开、登记或取消错误，由父工厂保原错并统一清理最高owner。472 """473 self.owner, self._budget = owner, owner.budget474 self.handle = self._digest = None475 self._fixed_cost = self._held_cost = self._count = 0476 self._last_time_ns, self._last_sequence = None, -1477 self.closed = self._closing = self.verified_eof = False478 # 父先接责任，register前后取消都不让真实FD或已经入账的费用脱离owner。479 owner._pending_reader = self480 # 新暂存允许8192字节有限预读；另预付16384覆盖buffer及对象，仍按逻辑行核限长/EOF。481 # 旧CAN读取不变；新暂存底层调用次数/取消时机可变化，handle仍负责buffer及raw FD。482 _reserve_owned_ram(self, 65536 + 64 * owner.max_row_bytes + 16384, '_fixed_cost')483 self._digest = sha256() if owner.integrity else None484 self.handle = owner.file_path.open('rb', buffering=8192)485 owner._readers.add(self)486487 def __iter__(self):488 """返回同一独立位置；参数: 无；返回: self；异常: 无。"""489 return self490491 def __next__(self):492 """限长读一行并核原序，完整EOF且成功close后才产生verified_eof。493494 参数: 无。返回: StoredDecodedMessage；最后一行交付仍未验证EOF。495 异常: StopIteration仅在关闭或真正完成；ResultFormatError为损坏；原IO、496 预算或取消错误保留并poison父，最高decoded_message_store_cleanup_owner可恢复。497 """498 if self.closed:499 raise StopIteration500 self.owner._require('sealed')501 if self._closing:502 raise ResultClosedError('解码仓reader正在清理')503 payload = record = None504 try:505 # 多读至多1字节识别超限；没有完整换行的尾行不能当作正常最后记录。506 limit = min(self.owner.max_row_bytes, self.owner.line_limit)507 payload = self.handle.readline(limit + 1)508 if payload:509 if len(payload) > limit or not payload.endswith(b'\n'):510 raise ResultFormatError('解码仓行超过封口实测上限/公共硬限或缺少换行')511 if self._count >= self.owner.count:512 raise ResultFormatError('解码仓封口后存在额外后缀')513 record = _decode_row(payload)514 if (record.decoded.sequence <= self._last_sequence if self.owner.sparse_sequences515 else record.decoded.sequence != self._count) or (516 self._last_time_ns is not None and record.time_ns < self._last_time_ns):517 raise ResultFormatError('解码仓原sequence或时间顺序损坏')518 if self._digest is not None:519 self._digest.update(payload)520 self._count += 1521 self._last_time_ns, self._last_sequence = record.time_ns, record.decoded.sequence522 return record523 # 真实EOF才核全流，不以已交付行数、最后一次readline或seal事实代替核验。524 if self._count != self.owner.count or self._digest is not None and self._digest.hexdigest() != self.owner.sha256:525 raise ResultFormatError('解码仓EOF条数或SHA不匹配')526 self.close()527 except BaseException as exc:528 payload = record = None529 self.owner._fail(exc)530 raise531 finally:532 # 调用方保留的异常或已返回记录不应使本reader继续保活已抵费单行副本。533 payload = record = None534 self.verified_eof = True535 raise StopIteration536537 def close(self):538 """只关闭本reader，可从持续真实FD或预算失败恢复而不影响其他位置。539540 参数: 无。返回: None；FD、费用及父登记都成功完成才closed=True。541 异常: 原FD、抵销、登记或取消错误，最高父owner可达；真实已关闭handle不重关，542 已抵销费用不重扣，提前close不设置verified_eof。543 """544 # 重试仅处理未完成责任，不能因先前部分成功再次关闭FD或抵销其他owner费用。545 if self.closed:546 return547 self._closing = True548 try:549 # 实际FD关闭前不解除父责任，失败后仍保留原handle供close重试。550 if self.handle is not None and not self.handle.closed:551 self.handle.close()552 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')553 if self.owner._readers is not None:554 self.owner._readers.discard(self)555 if self.owner._pending_reader is self:556 self.owner._pending_reader = None557 self.closed = True558 except BaseException as exc:559 exc.decoded_message_store_cleanup_owner = self.owner560 raise561