feat: custom indexer module (single index)

[whankinsiv]

Description

This PR introduces the initial custom indexer module. It enables downstream code to implement the ChainIndex trait and receive transaction events. The module currently supports a single index and a follow up PR will extend this to handle multiple indices within one module.

It also includes 2 CursorStore implementations (Fjall and in memory). Example usage is provided in processes/indexer/main.rs, including a simple pool cost index to demonstrate how to construct and run an indexer.

Related Issue(s)

Completes #375

How was this tested?

• Manually running the indexer process and verifying that state updates propagate through the watch channel.
• Restarting the process to confirm that the Fjall backed CursorStore restores the last persisted point and resumes sync from it.

Checklist

• My code builds and passes local tests
• I added/updated tests for my changes, where applicable
• I updated documentation (if applicable)
• CI is green for this PR

Impact / Side effects

No side effects for the omnibus process.

Reviewer notes / Areas to focus

custom_indexer.rs, especially the init flow and context.run behavior.

[SupernaviX]

Now that we're interfacing with code outside of acropolis itself, I think we need a new error-handling strategy. Specifically, if a custom index returns an error when processing a TX, we should not keep passing data to that index.

The reason for that is, the implementation of an index is coming from user code. If that user code throws an error, something is wrong which its code can't account for; maybe a datum is malformed, maybe their DB is down, maybe it's some bug. But the application's operator should be able to fix their code or infrastructure, and after they do, the application should resume indexing from wherever it left off.

[SupernaviX]

For similar reasons to the ones described above, if we can't decode a transaction, we should not apply future transactions. If someone submits a "malformed" transaction on-chain, we shouldn't just skip it; we should stop processing transactions until it gets rolled back or our decoder gets fixed.

[SupernaviX]

this'll be handled in a followup

[lowhung]

Not likely to be a problem here as I would expect the structure of Point to be stable (it's a simple two-variant enum), however for less stable types that you're encoding/decoding, I like to include a version prefix constant to ensure compatibility with previous versions of a struct's encoding/decoding. This is particularly relevant with bincode since it's a non-self-describing format—it doesn't store field names or type information, so changes to field order or structure can cause silent deserialization failures.

Something like the following:

const CURSOR_VERSION: u8 = 1;

async fn save(&self, point: &Point) -> Result<()> {
    let mut raw = vec![CURSOR_VERSION];
    raw.extend(bincode::serialize(point)?);
    self.cursor.insert("cursor", raw)?;
    Ok(())
}

async fn load(&self) -> Result<Option<Point>> {
    let Some(bytes) = self.cursor.get("cursor")? else {
        return Ok(None);
    };
    
    match bytes.first() {
        Some(1) => Ok(Some(bincode::deserialize(&bytes[1..])?)),
        Some(v) => anyhow::bail!("unsupported cursor version: {v}"),
        None => anyhow::bail!("empty cursor data"),
    }
}

Again, this is more pertinent to data that could be subject to change in the future.

[SupernaviX]

This could totally change in the future. A Point is just two fields, but as we index more forms of data we may need more info in these cursors.

[lowhung]

Makes sense 👍🏼

[SupernaviX]

this'll be handed in a followup

[whankinsiv]

Merging. Simon's feedback regarding how the indexer handles decode, rollback, and tx handling errors will be included in the follow up PR.