# `DdDataStreams.Pathway`
[🔗](https://github.com/rockneurotiko/dd_data_streams/blob/main/lib/dd_data_streams/pathway.ex#L1)

A pathway is the chain of nodes a payload passes through: the producer, the
broker edge, the consumer, and so on.

`hash` identifies the current node in that chain. It is a 64-bit FNV-1 hash of
the node's own identity (`node_hash/4`) chained onto the hash of the node that
sent the payload (`pathway_hash/2`), so the payload carries the same value on
to the next service. `pathway_start` and `edge_start` are the Unix times in
milliseconds at which the pathway and the current edge started; the
difference between them and the time of a checkpoint is the latency reported
for the edge. A `pathway_start` of `0` is the sentinel for an unstarted
pathway: its first checkpoint starts it at the checkpoint time.

Only tags that identify a node take part in the hash. Tags that vary per
message, such as `partition`, are reported but never hashed, because hashing
them would split one node into a node per partition and stop the downstream
produce edges from joining up.

`tags` are the edge tags of the node `hash` identifies, so the pathway a
consume left behind carries the consume edge's tags and a produce in that
stage can read them. They stay in the VM: `DdDataStreams.Propagator` writes
the hash and the two starts on the wire, not the tags, so a consumer cannot
read its producer's tags from the header it receives.

# `checkpoint`

```elixir
@type checkpoint() :: %{
  :hash => non_neg_integer(),
  :parent_hash => non_neg_integer(),
  :tags => [String.t()],
  :timestamp => integer(),
  :pathway_latency => non_neg_integer(),
  :edge_latency => non_neg_integer(),
  optional(:payload_size) => non_neg_integer(),
  optional(:service) => String.t()
}
```

The measurement a checkpoint produces. `pathway_latency` and `edge_latency`
are in milliseconds and measure back from the checkpoint's `timestamp`.
`payload_size` is the byte size of the message the checkpoint was recorded
for; `DdDataStreams.Checkpoints.consume/3` and `produce/3` add it for
aggregation, while `set_checkpoint/3` leaves it out and the aggregator
treats an absent value as zero. `service` is added the same way, and is the
service the checkpoint is reported under.

# `t`

```elixir
@type t() :: %DdDataStreams.Pathway{
  closest_opposite_direction_edge_start: term(),
  closest_opposite_direction_hash: term(),
  edge_start: non_neg_integer(),
  extractor: module() | nil,
  hash: non_neg_integer(),
  pathway_start: non_neg_integer(),
  previous_direction: term(),
  producer_info: DdDataStreams.Extractor.producer_info() | nil,
  tags: [String.t()]
}
```

# `merge`

```elixir
@spec merge([t()]) :: t()
```

Samples one pathway from a fan-in, so a batch that consumed several pathways
continues exactly one of them. An empty list yields an unstarted pathway.

# `new`

```elixir
@spec new(keyword() | map()) :: t()
```

Builds a pathway from the given fields, defaulting missing ones to zero.

    iex> DdDataStreams.Pathway.new(hash: 234)
    %DdDataStreams.Pathway{hash: 234, pathway_start: 0, edge_start: 0, extractor: nil}

    iex> DdDataStreams.Pathway.new(edge_start: 1700003600000)
    %DdDataStreams.Pathway{hash: 0, pathway_start: 0, edge_start: 1700003600000, extractor: nil}

# `node_hash`

```elixir
@spec node_hash(String.t(), String.t(), String.t() | nil, [String.t()]) ::
  non_neg_integer()
```

Returns the hash of the node identified by `service`, `env`, `primary_tag` and
`tags`.

The fields are concatenated without a separator: the base hash over service,
environment and primary tag, followed by the hashable edge tags in byte order.
Tags outside the hashable vocabulary are dropped; a tag repeated in `tags`
takes part as many times as it appears.

# `pathway_hash`

```elixir
@spec pathway_hash(non_neg_integer(), non_neg_integer()) :: non_neg_integer()
```

Chains a node hash onto the hash of the node that sent the payload.

The two hashes are hashed as 64-bit unsigned little-endian integers, node
first, so a downstream node can be computed from the parent it received.

# `set_checkpoint`

```elixir
@spec set_checkpoint(t(), [String.t()], keyword()) :: {t(), checkpoint()}
```

Sets a checkpoint for the node reached by `tags` at `:now` (milliseconds) and
returns the checkpointed pathway together with the measurement.

The first checkpoint of a pathway starts it at `:now`, or at the `:pathway_start`
override. Both starts can be overridden per checkpoint: the consume side uses
that to measure from the time the broker received the message rather than the
time the consumer picked it up.

Options:

  * `:now` - checkpoint time in milliseconds, defaults to the current time
  * `:service`, `:env`, `:primary_tag` - identity of the node; see `node_hash/4`
  * `:pathway_start`, `:edge_start` - override the respective start time

The returned pathway carries `tags` as the tags of the node it reached, so a
produce in the stage that consumed a message reads the consume edge's tags
from the pathway in `DdDataStreams.Context`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*