Dozer App State Machine

[chubei]

Dozer App State Machine

We only consider the single sink case now.

We intentionally don't distinguish between CacheSink and Cache because that should be an implementation detail irrelevant to this design.

State

The state of a node (source, processor, or sink) is a HashMap from source node handles to (u64, u64), where the first u64 is the transaction id, the second u64 the sequence number.

We say a state is None if the HashMap is empty, Some otherwise.

State is persisted to disk on commit. A node's state is persisted after all its parent nodes' states are persisted, so if a node has no persisted state, it's guaranteed that all its children have no persisted state.

Persisted states have following invariants:

• Is Some.
• The keys of the state of a child node is the union of the keys of the states of its parent nodes.
• A source state always has length 1 and the key is the source node handle.
• The keys of the sink state is all source node handles.
• The values of the sink state is less than or equal to the value of the state of corresponding source.

Consistency

A (sub)DAG is said to be consistent if and only if for every child node, its state is the union of its parent nodes' state.

We're going to test the consistency of the sub DAG that's generated by removing all the source nodes from the pipeline DAG. Let's call it sourceless DAG.

States

Dozer app has following states:

• Bootstrapping
• Snapshotting
• PopulatingProcessor
• PopulatingPipeline
• Streaming

When the app starts, it's always Bootstrapping.

State Transfer

Bootstrapping

Read the states of all nodes. If the persisted state can't be found, the node's state is None.

If any of the source states is None, transfer to Snapshotting.

At this point all source states are Some.

If the DAG is consistent, transfer to Streaming.

If the sourceless DAG is consistent, transfer to PopulatingPipeline.

Otherwise transfer to PopulatingProcessor.

Snapshotting

For every source node, if its state is None, run its snapshot method and persist state.
snapshot does what current source's start does when from_seq is None, but returns when the snapshotting is done.

Transfer to Bootstrapping.

PopulatingProcessor

Precondition: All sources states are Some.

Remove all processor's state.

If the sink state is None, the sourceless DAG is consistent now. Transfer to PopulatingPipeline.

Otherwise, construct execution DAG by creating source replay nodes, connecting them to processor nodes, and don't connect the sink node.
The replay from is set to None, to to the value of the sink state's corresponding value.

Run this execution DAG.

Postcondition: Sourceless DAG is consistent.

Transfer to PopulatingPipeline.

PopulatingPipeline

Precondition: All source states are Some. Sourceless DAG is consistent.

Construct execution DAG by creating source replay nodes, connecting them to the processor and sink nodes.
The replay from is set to the value of the sink state's corresponding value, to to the source state.

Run this execution DAG.

Postcondition: DAG is consistent.

Transfer to Streaming.

Streaming

Precondition: All source states are Some. DAG is consistent.

Construct execution DAG by creating source stream nodes, connecting them to the processor and sink nodes.
The stream from is set to the value of the source state.

Run this execution DAG, until stopped.

Source Replay

Transaction Log

Every source node maintains a Vec<TransactionLogEntry> as its state.

struct TransactionLogEntry {
  /// Exclusive.
  from: Option<(u64, u64)>,
  /// Inclusive.
  to: (u64, u64),
  /// Data.
  ops: Vec<Operation>,
}

The persisted source state's only value is the last entry's to value.

The transaction log entries have following invariants:

assert_eq!(entries.first().map(|entry| entry.from).unwrap_or(None), None)

for i in 0..entries.len() - 1 {
  assert_eq!(entries[i + 1].from, Some(entries[i].to))
}

If the sink state is Some, there exists an entry such that entry.to equals to the value of the sink state's corresponding value.

Replay

Using the transaction log, we can replay the source node's output operations between states. A source replay node in an execution DAG is defined as following:

struct SourceReplayNode<'a> {
    /// The relevant transaction log.
    entries: &'a [TransactionLogEntry],
}

impl<'a> SourceReplayNode<'a> {
    pub fn new(whole_log: &'a [TransactionLogEntry], from: Option<(u64, u64)>, to: (u64, u64)) -> Self {
        todo!()
    }
}

When started, the node emits the operations in the transaction log entries one by one.

Source Stream

What current source's start does when from_seq is Some.

@v3g42 @mediuminvader @karolisg @snork-alt

[snork-alt]

Why must each source maintain a Vec<TransactionLogEntry> ? and not a single TransactionLogEntry? Does the theVec<Operation> of the TransactionLogEntry keep the entire history or only part of it?

I do not think we need to keep the whole history. It should be enough to maintain all the operations with a (tx, seq_no) higher than the one present in the cache to be able to align the cache to the source state. The pipeline state can be reconstructed from the RecordWriter state, which currently maintains a snapshot of all data. It should be enough to maintain in the RecordWriter a history of messages needed to re-align the cache.

Let's take an example. Let's say cache is at position 5 and RecordWriter is at position 7. This means RecordWriter has all records as of 5, and maintains a partial history from 5 to 7. If Dozer is resumed in this state, I can reconstruct the state of the pipeline by streaming the snapshot (as of 5) from RecordWriter as Insert operations. Once the pipeline is at state 5, I can then start streaming all operations from 5 to 7, so that both the pipeline and cache will align to 7. After that, I start streaming from the source (Postgres) with start_seq_no = 7.

[chubei]

In this example, why the RecordWriter conveniently has 5 and 5-7 instead of 6 and 6-7. In other words, how does RecordWriter guarantee it's consistent with cache?

[chubei]

We need a Vec of TransactionLogEntries because we may need to populate processor until souceless DAG is consistent, but we can't predict the sink state when committing a TransactionLogEntry.

Indeed we don't need the whole history (which is not in the TransactionLogEntry invariants), we only need to guarantee: 1. We can replay from None to sink state. 2. We can replay from sink state to source state.

[chubei]

Continuing this discussion in #771