Node state passing

[b3by]

Description

This PR offers an implementation of a node state passing mechanism.

State object
The new State object extends UserDict, so it can be used like any regular dictionary. Additionally, it holds a slim history of deltas, that can be used to recreate a state. Once retrieved, deltas are deleted from the state object as to prevent excessive duplication.

>>> import juturna as jt
>>> s = jt.components.State()
>>> s['a'] = 10
>>> s['b'] = 20
>>> s._ops
{'a': ('SET', 10), 'b': ('SET', 20)}
>>> d = s.deltas()
>>> d
[('SET', 'a', 10), ('SET', 'b', 20)]
>>> s._ops
{}
>>> s['c'] = 30
>>> s._ops
{'c': ('SET', 30)}
>>> del s['c']
>>> s._ops
{'yolo': ('DEL', None)}

Using deltas instead of full states allows, if needed, to share only relevant changes about a node state, thus avoiding retransmission in performance-critical scenarios.

Nodes as fully stateless entities
Nodes receive an external state object reference, and use that reference internally whenever the update method is called. The method signature now becomes:

def update(self, message: Message[T_Input], state: State): ...

Nodes still retain the option to use dirty state, as well as to use their state values through self._state.

State store in the pipeline
The pipeline object now holds a dictionary where node names are the keys, and states are the values. Whenever a new node is instantiated in a pipeline, its state is created and linked with node.link_state(state_store[node_name].

Warped nodes
Similarly to what the pipeline object does, the remote service creates a state store, but in this case the dictionary does not hold node names as keys but pipe ids. This solves the issue of a warped node shared across multiple pipes: when this happens, every time a new message is available from a pipe, the service links the correct state on the node and then pushes the message to its queue.

This PR also includes changes for all the built-in nodes.

PR type
Select all the labels that apply to this PR.

• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Code refactoring
• Performance improvement

Affected components

• Base node
• Pipeline
• Remot service
• Warp node

[GaijinKa]

This is a huuuuuuge change.

I'm ok with the pipeline control of the state, and with the capability of using a dirty state.

The update change is a big thing: once we moved to the new signature, ALL the existing nodes should be bumped, even the ones that do not require state. I'm wondering if we can/should add a syntactic sugar for this, but this is not a stopping issue for the merge at all.

OTOH, I have different concerns about the deltas() approach.

I'd like to hold off on merging this (I'm talkin of the deltas part). Not because the delta-based design is wrong in principle, but because there's currently no state-passing mechanism at all to compare it against, and we're taking on a breaking change across the base node, pipeline, remote service, and warp nodes to introduce one.

In a nutshell, aside the good-qualiy code you wrote:
This is effectively a premature optimization, that penalizes the system's resiliency, failover, and horizontal scalability by introducing risks of silent data drift and forcing stateful node affinity.

We're choosing the first implementation here, not optimizing an existing one, so I think the choice deserves more scrutiny.

Pros of the delta approach (SET/DEL) as implemented

• Bandwidth proportional to the actual change, not the full state size — though we don't have a case where payload size vs. update rate is actually a known constraint. What data are we sending here that makes this ratio critical? Without that, this is a theoretical benefit rather than one we can measure.
• deltas() gives a clean, purge-on-read history without manual diffing.

Cons / risks

• Ordering and delivery guarantees: _ops is coupled to the sequence of calls on this instance. A missed deltas() call, an out-of-order read, or a delta dropped in transit leaves the remote replica silently drifted from the source of truth - there's no sequence number or checksum to detect or recover from this.

• No resync path: deltas() clears _ops on read, so there's no way for a consumer to rebuild state from scratch (new subscriber, reconnect after failure) without a full-state fallback.

• Multi-node / warp node scenario: the warp node should be shared across multiple pipes via a state store keyed by pipe id - exactly the case where delta-based sync is fragile, since interleaved updates or a service restart have no described way to detect divergence or force a resync.

• Variable-size state in JSON format:
  I'm not so sure about how this should work with arrays and nested elements.
  State encapsulates UserDict and can contain nested values ​​of arbitrary shape/size. The higher-level SET/DEL handles only simple substitution;
  it does not generalize to nested comparisons (array index instability, need for path-based patches) if necessary. This severely penalizes the solution in real-world scenarios.
  If I understood, I think that to make it worth on big object like the YOLO results, an approach similar to that of JSON Patch (RFC 6902) should be considered.

• Scalability and failover: Full state updates are autonomous, so any node can handle any request: a load balancer can be placed upstream without affinity requirements, and failover simply means the next node sends the next full state. This is a critical requirement!
  Delta-based synchronization requires the receiving/generating side to keep track of the sequence state for each client, which implies node affinity and -as already said - a failover path that must still retrieve the latest full state.

IMHO, I'd like to start with the simplest model: implement the send the full state on each update in warp:
It avoids all the drift/resync/affinity issues above, is trivial to reason about and debug, and gives us a working baseline to observe real usage against - actual payload sizes, update frequency, and whether bandwidth or scaling ever become real constraints. If profiling or real scenarios later show full-state is a bottleneck, that's the point to introduce delta-based transfer, informed by actual numbers instead of upfront.

Given the size of the surface this touches, I'd rather land the simpler version first and let real needs drive the added complexity later. Open to revisiting if there's a specific workload or constraint behind the delta design that I'm missing.

[b3by]

I applied the changes and fixed the issues with the warp nodes.

Just to make a quick recap of the key points of this PR that were not included in the original text:

• node state is implemented with a class; albeit not needed now, this was done so that in the future, road will already be paved in case more logic should be added to it (a basic example is the deltas, which are now not included in the class);
• communication between local pipes and remote service always includes the full state in the enveloped messages transmitted and received, so local node passes the state to the remote, and the remote sends the state back along with the response; again, this mechanism is the simplest at the moment, and there are no strict requirements or use cases that suggest this could become a bottleneck, especially considering states are often objects of very modest size;
• the update signature now includes **kwargs, so everything that is added to the update call can be added there without the need to change existing implementations (for this particular PR, `kwargs['state'] contains the state passed when a new message is received).