Lightning.Adaptors.Scheduler (Lightning v2.19.0)

View Source

Cluster-singleton GenServer that refreshes the catalogue from the strategy, on a timer and on demand, persisting through Lightning.Adaptors.Catalogue and broadcasting {:changed, name, source} on the source topic.

The first tick is due one refresh_interval after the catalogue's most recent check, so a restart does not refetch straight away; an empty catalogue ticks at once. An interval of 0 disables the timer and leaves only on-demand refreshes.

A tick lists the source and fetches every adaptor whose latest_version changed. It also refetches an adaptor whose stored row has no schema, for the grace period set by @schema_grace_ms after the row's updated_at. jsDelivr mirrors a new version with some lag, so a schema missing inside that window may still arrive. After it, a missing schema is taken as really missing. A refetch that still finds no schema counts as touched. Icons are fetched in parallel and each changed adaptor is upserted with its icons. refresh_package/2 refetches one adaptor without icons.

Summary

Types

refresh_counts()

One refresh cycle's tallies: adaptors the upstream listing returned, how many of those had a changed latest_version, how many were fetched and persisted, and how many adaptors the cycle failed to fetch or to write.

Functions

await_refresh(scheduler_name, timeout)

Starts a refresh cycle, or joins the one in flight, and waits for it to complete.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

completed?(scheduler_name)

Whether a cycle has completed against a source that listed no adaptors at all, since this Scheduler started.

refresh_icons(scheduler_name)

Refetches every adaptor's icons and updates the rows whose icon bytes changed, leaving other fields untouched.

refresh_now(scheduler_name)

Starts a refresh tick, or lets one already in flight continue.

refresh_package(scheduler_name, name)

Refetches and persists one adaptor whether or not its version changed. Waits up to 30 seconds.

start_link(opts)

Starts the Scheduler. Required opts: :name, :sup, :lock_key, :cache, :tasks, :source_topic, :refresh_interval (tick interval in milliseconds; 0 disables the timer) and :warn_when_empty (whether booting on an empty catalogue with the timer disabled logs a warning). :checked_at is optional. It is a 1-arity function, defaulting to &Catalogue.max_checked_at/1, that reads the source's last-checked timestamp. It is called once at boot to work out the delay before the first tick.

Types

refresh_counts()

@type refresh_counts() :: %{
  listed: non_neg_integer(),
  changed: non_neg_integer(),
  fetched: non_neg_integer(),
  errors: non_neg_integer()
}

One refresh cycle's tallies: adaptors the upstream listing returned, how many of those had a changed latest_version, how many were fetched and persisted, and how many adaptors the cycle failed to fetch or to write.

Functions

await_refresh(scheduler_name, timeout)

@spec await_refresh(GenServer.server(), timeout()) ::
  {:ok, refresh_counts()} | {:error, {:refresh_failed, term()} | term()}

Starts a refresh cycle, or joins the one in flight, and waits for it to complete.

Returns {:ok, counts} when the listing succeeded (per-adaptor fetch and upsert failures are counted in counts.errors), {:error, reason} when it failed, or {:error, {:refresh_failed, reason}} when the cycle crashed.

A caller whose timeout expires before the cycle finishes is dropped rather than replied to, so a late result never lands in its mailbox.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

completed?(scheduler_name)

@spec completed?(GenServer.server()) :: boolean()

Whether a cycle has completed against a source that listed no adaptors at all, since this Scheduler started.

That is the one outcome no row can record. An upstream answering with an empty list has told us there are no adaptors, and the empty catalogue it leaves behind is loaded rather than unloaded. Every other completed cycle leaves rows, which answer for themselves and keep answering after a restart. So this deliberately says nothing about them, and a source whose rows are later deleted reloads as it did before.

A cycle that failed to list, fetch or write is not a completed one. We cannot tell a source with nothing in it from one we could not read.

Answers false for a Scheduler that is unreachable.

refresh_icons(scheduler_name)

@spec refresh_icons(GenServer.server()) ::
  {:ok, %{updated: non_neg_integer(), unchanged: non_neg_integer()}}
  | {:error, term()}

Refetches every adaptor's icons and updates the rows whose icon bytes changed, leaving other fields untouched.

Returns {:ok, %{updated: n, unchanged: m}}, {:error, reason} if the fetch fails, or {:error, {:refresh_failed, reason}} if the task crashed.

refresh_now(scheduler_name)

@spec refresh_now(GenServer.server()) :: :ok | {:error, term()}

Starts a refresh tick, or lets one already in flight continue.

refresh_package(scheduler_name, name)

@spec refresh_package(GenServer.server(), String.t()) ::
  :ok | {:error, :not_found | term()}

Refetches and persists one adaptor whether or not its version changed. Waits up to 30 seconds.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts the Scheduler. Required opts: :name, :sup, :lock_key, :cache, :tasks, :source_topic, :refresh_interval (tick interval in milliseconds; 0 disables the timer) and :warn_when_empty (whether booting on an empty catalogue with the timer disabled logs a warning). :checked_at is optional. It is a 1-arity function, defaulting to &Catalogue.max_checked_at/1, that reads the source's last-checked timestamp. It is called once at boot to work out the delay before the first tick.