Repository: Chia-Network/chia-blockchain
File: chia/data_layer/data_store.py (Python)

Code before:
                self.unconfirmed_keys_values[store_id].append(blob_hash)
        except sqlite3.IntegrityError as e:
            if "UNIQUE constraint failed" in str(e):
                kv_id = await self.get_kvid(blob, store_id)
                if kv_id is None:
                    raise Exception("Internal error") from e
                return kv_id

            raise

        if row is None:
            raise Exception("Internal error")
        if use_file:
            path = self.get_key_value_path(store_id=store_id, blob_hash=blob_hash)
            path.parent.mkdir(parents=True, exist_ok=True)
            # TODO: consider file-system based locking of either the file or the store directory
            path.write_bytes(zstd.compress(blob))
        return KeyOrValueId(row[0])

    async def delete_unconfirmed_kvids(self, store_id: bytes32) -> None:
        for blob_hash in self.unconfirmed_keys_values[store_id]:
            path = self.get_key_value_path(store_id=store_id, blob_hash=blob_hash)
            try:
                path.unlink()
            except FileNotFoundError:
                log.error(f"Cannot find key/value path {path} for hash {blob_hash}")
        self.unconfirmed_keys_values[store_id].clear()

    async def confirm_all_kvids(self, store_id: bytes32) -> None:
        self.unconfirmed_keys_values[store_id].clear()

    async def add_key_value(
        self, key: bytes, value: bytes, store_id: bytes32, writer: aiosqlite.Connection, confirmed: bool = True
    ) -> tuple[KeyId, ValueId]:
        kid = KeyId(await self.add_kvid(key, store_id, writer=writer, confirmed=confirmed))
        vid = ValueId(await self.add_kvid(value, store_id, writer=writer, confirmed=confirmed))

        return (kid, vid)

    async def get_terminal_node_by_hash(
        self,
        node_hash: bytes32,
        store_id: bytes32,
        root_hash: Union[bytes32, Unspecified] = unspecified,
    ) -> TerminalNode:
        resolved_root_hash: Optional[bytes32]
        if root_hash is unspecified:
            root = await self.get_tree_root(store_id=store_id)
            resolved_root_hash = root.node_hash
        else:
            resolved_root_hash = root_hash

Review comment (bug):
i...  think in this case it's reasonable to attempt to continue after logging regardless of the failure.  but always good to question ideas like this...

```suggestion
            with log_exceptions(log=log, consume=True):
                path = self.get_key_value_path(store_id=store_id, blob_hash=blob_hash)
                try:
                    path.unlink()
                except FileNotFoundError:
                    log.error(f"Cannot find key/value path {path} for hash {blob_hash}")
```

Code after:

        if row is None:
            raise Exception("Internal error")
        if use_file:
            path = self.get_key_value_path(store_id=store_id, blob_hash=blob_hash)
            path.parent.mkdir(parents=True, exist_ok=True)
            self.unconfirmed_keys_values[store_id].append(blob_hash)
            # TODO: consider file-system based locking of either the file or the store directory
            path.write_bytes(zstd.compress(blob))
        return KeyOrValueId(row[0])

    def delete_unconfirmed_kvids(self, store_id: bytes32) -> None:
        for blob_hash in self.unconfirmed_keys_values[store_id]:
            with log_exceptions(log=log, consume=True):
                path = self.get_key_value_path(store_id=store_id, blob_hash=blob_hash)
                try:
                    path.unlink()
                except FileNotFoundError:
                    log.error(f"Cannot find key/value path {path} for hash {blob_hash}")
        del self.unconfirmed_keys_values[store_id]

    def confirm_all_kvids(self, store_id: bytes32) -> None:
        del self.unconfirmed_keys_values[store_id]

    @contextmanager
    def manage_kv_files(self, store_id: bytes32) -> Iterator[None]:
        if store_id not in self.unconfirmed_keys_values:
            self.unconfirmed_keys_values[store_id] = []
        else:
            raise Exception("Internal error: unconfirmed keys values cache not cleaned")

        try:
            yield
        except:
            self.delete_unconfirmed_kvids(store_id)
            raise
        else:
            self.confirm_all_kvids(store_id)

    async def add_key_value(
        self, key: bytes, value: bytes, store_id: bytes32, writer: aiosqlite.Connection
    ) -> tuple[KeyId, ValueId]:
        kid = KeyId(await self.add_kvid(key, store_id, writer=writer))
        vid = ValueId(await self.add_kvid(value, store_id, writer=writer))

        return (kid, vid)

    async def get_terminal_node_by_hash(
        self,
        node_hash: bytes32,
        store_id: bytes32,