Persistent config

config/config.exs

@@ -19,6 +19,9 @@ config :mime, :types, %{
   "text/plain" => ["livemd"]
 }
 
+# Sets the default storage backend
+config :livebook, :storage, Livebook.Storage.Ets
+
 # Sets the default authentication mode to token
 config :livebook, :authentication_mode, :token
 

lib/livebook/application.ex

@@ -31,7 +31,9 @@ defmodule Livebook.Application do
         # Start the Endpoint (http/https)
         # We skip the access url as we do our own logging below
         {LivebookWeb.Endpoint, log_access_url: false}
-      ] ++ app_specs()
+      ] ++
+        app_specs() ++
+        storage_specs()
 
     opts = [strategy: :one_for_one, name: Livebook.Supervisor]
 
@@ -167,4 +169,12 @@ defmodule Livebook.Application do
   else
     defp app_specs, do: []
   end
+
+  defp storage_specs() do
+    if Mix.env() == :test do
+      []
+    else
+      [Application.fetch_env!(:livebook, :storage)]
+    end
+  end
 end

lib/livebook/storage.ex

@@ -0,0 +1,41 @@
+defmodule Livebook.Storage do
+  @moduledoc """
+  Behaviour defining an interface for storing arbitrary data in
+  [Entity-Attribute-Value](https://en.wikipedia.org/wiki/Entity%E2%80%93attribute%E2%80%93value_model) fashion.
+  """
+
+  @type namespace :: atom()
+  @type entity_id :: binary()
+  @type attribute :: atom()
+  @type value :: binary()
+  @type timestamp :: non_neg_integer()
+
+  @type entity :: %{required(:id) => entity_id(), optional(attribute()) => value()}
+
+  @doc """
+  Returns a map identified by `entity_id` in `namespace`.
+      fetch(:filesystem, "rand-id")
+      #=> {:ok, %{id: "rand-id", type: "s3", bucket_url: "/...", secret: "abc", access_key: "xyz"}}
+
+  """
+  @callback fetch(namespace(), entity_id()) :: {:ok, entity()} | :error
+
+  @doc """
+  Returns all values in namespace.
+
+      all(:filesystem)
+      [%{id: "rand-id", type: "s3", bucket_url: "/...", secret: "abc", access_key: "xyz"}]
+
+  """
+  @callback all(namespace()) :: [entity()]
+
+  @doc """
+  Inserts given list of attribute-value paris to a entity belonging to specified namespace.
+  """
+  @callback insert(namespace(), entity_id(), [{attribute(), value()}]) :: entity()
+
+  @doc """
+  Deletes an entity of given id from given namespace.
+  """
+  @callback delete(namespace(), entity_id()) :: :ok
+end

lib/livebook/storage/ets.ex

@@ -0,0 +1,110 @@
+defmodule Livebook.Storage.Ets do
+  @moduledoc """
+  Ets implementation of `Livebook.Storage` behaviour.
+
+  The module is supposed to be started just once as it
+  is responsible for managing a single global named ets table.
+
+  `insert` and `delete` operations are supposed to be called using the genserver
+  while all the lookups can be performed by directly accessing the named table.
+  """
+  @behaviour Livebook.Storage
+
+  @table_name :__livebook_storage__
+
+  use GenServer
+
+  @impl Livebook.Storage
+  def fetch(namespace, entity_id) do
+    @table_name
+    |> :ets.lookup({namespace, entity_id})
+    |> case do
+      [] ->
+        :error
+
+      entries ->
+        entries
+        |> Enum.map(fn {_key, attr, val, _timestamp} ->
+          {attr, val}
+        end)
+        |> Map.new()
+        |> Map.put(:id, entity_id)
+        |> then(&{:ok, &1})
+    end
+  end
+
+  @impl Livebook.Storage
+  def all(namespace) do
+    @table_name
+    |> :ets.match({{namespace, :"$1"}, :"$2", :"$3", :_})
+    |> Enum.group_by(
+      fn [entity_id, _attr, _val] -> entity_id end,
+      fn [_id, attr, val] -> {attr, val} end
+    )
+    |> Enum.map(fn {entity_id, attributes} ->
+      attributes
+      |> Map.new()
+      |> Map.put(:id, entity_id)
+    end)
+  end
+
+  @impl Livebook.Storage
+  def insert(namespace, entity_id, attributes) do
+    GenServer.call(__MODULE__, {:insert, namespace, entity_id, attributes})
+  end
+
+  @impl Livebook.Storage
+  def delete(namespace, entity_id) do
+    GenServer.call(__MODULE__, {:delete, namespace, entity_id})
+  end
+
+  @spec start_link() :: GenServer.on_start()
+  def start_link() do
+    GenServer.start_link(__MODULE__, [], name: __MODULE__)
+  end
+
+  @impl GenServer
+  def init(_opts) do
+    table = :ets.new(@table_name, [:named_table, :protected, :duplicate_bag])
+
+    {:ok, %{table: table}}
+  end
+
+  @impl GenServer
+  def handle_call({:insert, namespace, entity_id, attributes}, _from, %{table: table} = state) do
+    new_attributes =
+      attributes
+      |> Map.new(fn {key, value} ->
+        {key, {value, System.os_time(:millisecond)}}
+      end)
+
+    old_attributes =
+      table
+      |> :ets.lookup({{namespace, entity_id}, :"$1", :"$2", :"$3"})
+      |> Map.new(fn [key, value, timestamp] ->
+        {key, {value, timestamp}}
+      end)
+
+    # delete the whole entity to insert all its entries once again
+    # it is due to a limitation of duplicate bag where a single value can't be deleted nor updated
+    # therefore we are basically reinserting all attributes
+    :ets.delete(table, {namespace, entity_id})
+
+    old_attributes
+    |> Map.merge(new_attributes)
+    |> Enum.map(fn {key, {value, timestamp}} ->
+      :ets.insert(table, {{namespace, entity_id}, key, value, timestamp})
+    end)
+
+    {:ok, entity} = fetch(namespace, entity_id)
+
+    {:reply, entity, state}
+  end
+
+  @impl GenServer
+  def handle_call({:delete, namespace, entity_id}, _from, %{table: table} = state) do
+    :ets.delete(table, {namespace, entity_id})
+
+    {:reply, :ok, state}
+  end
+end

test/livebook/storage/ets_test.exs

@@ -0,0 +1,63 @@
+defmodule Livebook.Storage.EtsTest do
+  use ExUnit.Case, async: false
+
+  alias Livebook.Storage.Ets
+
+  setup_all do
+    {:ok, _pid} = Ets.start_link()
+
+    []
+  end
+
+  describe "insert/3" do
+    test "properly inserts a new key-value attribute" do
+      assert %{
+               id: "test",
+               key1: "val1",
+               key2: "val2"
+             } = Ets.insert(:insert, "test", key1: "val1", key2: "val2")
+    end
+
+    test "replaces already existing attributes with new values" do
+      assert %{
+               key1: "val1",
+               key2: "val2"
+             } = Ets.insert(:insert, "replace", key1: "val1", key2: "val2")
+
+      assert %{
+               key1: "updated_val1",
+               key2: "val2",
+               key3: "val3"
+             } = Ets.insert(:insert, "replace", key1: "updated_val1", key2: "val2", key3: "val3")
+    end
+  end
+
+  test "fetch/2" do
+    _entity = Ets.insert(:fetch, "test", key1: "val1")
+
+    assert {:ok,
+            %{
+              id: "test",
+              key1: "val1"
+            }} = Ets.fetch(:fetch, "test")
+
+    assert :error = Ets.fetch(:fetch, "unknown")
+  end
+
+  test "delete/2" do
+    _entity = Ets.insert(:delete, "test", key1: "val1")
+
+    assert {:ok, _entity} = Ets.fetch(:delete, "test")
+
+    assert :ok = Ets.delete(:delete, "test")
+
+    assert :error = Ets.fetch(:delete, "test")
+  end
+
+  test "all/1" do
+    entity1 = Ets.insert(:all, "test1", key1: "val1")
+    entity2 = Ets.insert(:all, "test2", key1: "val1")
+
+    assert [^entity1, ^entity2] = Ets.all(:all)
+  end
+end