# `Ankusa.Queue.Syncer`
[🔗](https://github.com/jamescarr/ankusa/blob/ankusa-v0.7.0/packages/ankusa/lib/ankusa/queue/syncer.ex#L1)

The fsync of the queue's WAL, off the writer's critical path (DUR-4.b).

`Ankusa.Queue.Writer` applies a group with `sync: false`, which makes it
visible to every reader at once, and hands what must wait for the disk to this
process: the replies of the group's callers (`{:reply, calls, reply}`), or, under
`wal: :dual`, the writer's own notice (`{:notify, pid, ref}`) that the local
half is durable. The writer then assembles the next group while this process
syncs the one before it.

One `Ankusa.Store.sync_wal/1` is in flight at a time. It takes every entry
queued behind it and answers them all once the fsync returns, so the fsyncs
are as many as the disk allows, not one per group. `sync_wal` does not enter
RocksDB's write queue, so the writer's next `sync: false` write is applied
meanwhile.

## The durable watermark

Three counters (`marks`, an `:atomics` array the writer registers as
`{instance, :queue_marks}`):

  * `durable_seq`: the highest seq every group up to which has been fsynced;
  * `applied_seq`: the highest seq the writer has applied (set *before* the
    write, so a reader that sees the batch also sees it as not yet durable);
  * `durable_at`: the stamp (`now` of the writer) of the newest durable group.

A row is externally visible or binding only once it is durable (ground rules 1
and 5): `visible/2` bounds what the dispatch pipeline may scan while a group is
applied but not synced, the archive scan stops at `durable_seq`, and a reply is
never sent before the sync that covers every seq it depends on (a `201
duplicate` for a key written by a group still in the pipeline waits for that
group's fsync too). Only the ack latency is pipelined: the visibility of a hook
and the answer to its retry are not.

A failed sync **fences** the queue. This process answers every entry it covers,
and every one that arrives after, with the error, leaves `durable_seq` where it
was and does not sync again; the store has been told
(`Store.report_write_failure/2`, which reopens it) and the writer is told, which
refuses new work, deletes in one synced write the rows of every group that was
never acked, and only then sends `{:resume, seq, at}`, which raises the
watermark to where the writer stands. So a hook whose provider was told `503`
is never visible to dispatch (it is deleted before the watermark moves) and is
not found by the provider's retry: it is stored once, by the retry.

# `entry`

```elixir
@type entry() :: %{
  upto: non_neg_integer(),
  at: integer(),
  done: {:reply, term(), term()} | {:notify, pid(), term()}
}
```

# `marks`

```elixir
@type marks() :: :atomics.atomics_ref()
```

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `durable_seq`

```elixir
@spec durable_seq(atom()) :: non_neg_integer() | :infinity
```

The seq the archive may read up to: every row at or below it is durable.

# `mark_durable`

```elixir
@spec mark_durable(marks(), non_neg_integer(), integer()) :: :ok
```

Everything up to `seq` (stamp `at`) is durable: raises the watermark, never
lowers it. The writer calls it after a `sync: true` write, which fsyncs the
whole WAL.

# `resume`

```elixir
@spec resume(pid(), non_neg_integer(), integer()) :: :ok
```

The writer has deleted the rows of the groups a failed fsync left: every seq up
to `seq` is settled (stamp `at`), and the syncer syncs again.

# `submit`

```elixir
@spec submit(pid(), entry()) :: :ok
```

Queue `entry` for the next fsync; answered when one covers it.

# `unsynced?`

```elixir
@spec unsynced?(marks()) :: boolean()
```

Whether a group is applied that no fsync has covered yet.

# `visible`

```elixir
@spec visible(atom(), integer()) :: integer()
```

The latest time dispatch may treat as "now": `now` while every applied group is
durable, else the stamp of the newest durable group less one millisecond, so a
row of a group still in the pipeline (stamped at or after it) is never due.

---

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