Speed up forge blob pages

Diff

8 files changed, +208 -25

    end
  end

  @doc "Read the data for a file page after one cache-freshness check."
  def blob_page(repo, ref, path) do
    with :ok <- check(repo, ref),
         :ok <- check_path(path) do
      _ = freshen(repo)

      with {:ok, sha} <- resolve_commit_from_cache(repo, ref),
           {:ok, blob} <- blob_from_cache(repo, sha, path) do
        head = with {:ok, current} <- head_from_cache(repo), do: current
        {:ok, %{sha: sha, head: head, blob: blob}}
      end
    end
  end

  defp blob_from_cache(repo, ref, path) do
    with {:ok, full} <- resolve_commit_from_cache(repo, ref) do
      spec = full <> ":" <> path

53	53		`do: {:reply, %{"ready" => map_size(failures) == 0, "failures" => failures}, failures}`
54	54
55	55		`def handle_call(:reset, _from, _failures), do: {:reply, :ok, %{}}`
	56	+
	57	+	`@impl true`
	58	+	`def handle_info(`
	59	+	`{:"ETS-TRANSFER", OpenAgents.Forge.WAL.Gcs.IndexCache, _owner, :wal_index_cache},`
	60	+	`failures`
	61	+	`),`
	62	+	`do: {:noreply, failures}`
56	63		`end`

283	283		`[stderr_to_stdout: true] ++ opts`
284	284		`)`
285	285		`end`
	286	+
	287	+	`@doc false`
	288	+	`def git_with_stdin(git_dir, args, input, opts \\ []) when is_binary(input) do`
	289	+	`input_path =`
	290	+	`Path.join(`
	291	+	`System.tmp_dir!(),`
	292	+	`"forge-git-input-#{System.unique_integer([:positive])}-#{:erlang.phash2(self())}"`
	293	+	`)`
	294	+
	295	+	`File.write!(input_path, input)`
	296	+
	297	+	`try do`
	298	+	`System.cmd(`
	299	+	`"sh",`
	300	+	`[`
	301	+	`"-c",`
	302	+	`~s(exec git "$@" < "$OPENAGENTS_GIT_INPUT"),`
	303	+	`"sh",`
	304	+	`"--git-dir",`
	305	+	`git_dir \| args`
	306	+	`],`
	307	+	`[env: [{"OPENAGENTS_GIT_INPUT", input_path}], stderr_to_stdout: true] ++ opts`
	308	+	`)`
	309	+	`after`
	310	+	`File.rm(input_path)`
	311	+	`end`
	312	+	`end`
286	313		`end`

  end

  defp refs_materialized_at?(path, index) do
    index
    |> WAL.refs()
    |> Map.values()
    |> Enum.uniq()
    |> Enum.all?(fn sha ->
      match?({_output, 0}, Repos.git(path, ["cat-file", "-e", sha]))
    end)
    tips = index |> WAL.refs() |> Map.values() |> Enum.uniq()

    case tips do
      [] ->
        true

      tips ->
        input = Enum.join(tips, "\n") <> "\n"

        case Repos.git_with_stdin(path, ["cat-file", "--batch-check"], input) do
          {output, 0} ->
            lines = String.split(output, "\n", trim: true)

            length(lines) == length(tips) and
              Enum.all?(lines, &(not String.ends_with?(&1, " missing")))

          _error ->
            false
        end
    end
  end

  # Apply one WAL entry: materialize its objects, prove the objects it

  @storage_base "https://storage.googleapis.com"
  @metadata_token_url "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token"
  @token_cache_key {__MODULE__, :token}
  @index_cache_table OpenAgents.Forge.WAL.Gcs.IndexCache
  @token_expiry_margin_seconds 60
  @stream_chunk_bytes 1_048_576
  @default_stream_timeout_ms 6 * 60 * 60 * 1_000

    with {:ok, bucket} <- bucket() do
      name = index_object(repo)

      with {:ok, generation} <- fetch_generation(bucket, name),
           {:ok, raw} <- download(bucket, name),
           {:ok, index} <- decode_index(raw) do
        {:ok, generation, index}
      with {:ok, generation} <- fetch_generation(bucket, name) do
        case cached_index(repo, generation) do
          {:ok, index} ->
            {:ok, generation, index}

          :miss ->
            with {:ok, raw} <- download(bucket, name),
                 {:ok, index} <- decode_index(raw) do
              cache_index(repo, generation, index)
              {:ok, generation, index}
            end
        end
      end
    end
  end

             headers: auth_headers(token) ++ [{"content-type", "application/json"}]
           ) do
        {:ok, %Req.Response{status: 200, body: %{"generation" => generation}}} ->
          cache_index(repo, generation, index)
          {:ok, generation}

        {:ok, %Req.Response{status: 412}} ->

  def delete_repo(repo) do
    with {:ok, bucket} <- bucket(),
         {:ok, token} <- token() do
      delete_prefix(bucket, prefix(repo), token)
      case delete_prefix(bucket, prefix(repo), token) do
        :ok ->
          delete_cached_index(repo)
          :ok

        {:error, _reason} = error ->
          error
      end
    end
  end

  @doc false
  def reset_index_cache do
    case :ets.whereis(@index_cache_table) do
      :undefined -> :ok
      _table -> :ets.delete_all_objects(@index_cache_table)
    end

    :ok
  end

  ## Object naming (public so it is testable without a live bucket)

  @doc """

    with {:ok, token} <- token() do
      url = object_url(bucket, name) <> "?" <> URI.encode_query(fields: "generation")

      case Req.get(url, headers: auth_headers(token)) do
      case Req.get(url, request_options(headers: auth_headers(token))) do
        {:ok, %Req.Response{status: 200, body: %{"generation" => generation}}} ->
          {:ok, generation}


    with {:ok, token} <- token() do
      url = object_url(bucket, name) <> "?alt=media"

      case Req.get(url, headers: auth_headers(token), decode_body: false) do
      case Req.get(
             url,
             request_options(headers: auth_headers(token), decode_body: false)
           ) do
        {:ok, %Req.Response{status: 200, body: body}} -> {:ok, body}
        {:ok, %Req.Response{status: 404}} -> {:error, :not_found}
        {:ok, %Req.Response{status: status, body: body}} -> {:error, {:gcs_error, status, body}}

261	291		`end`
262	292		`end`
263	293
	294	+	`defp cached_index(repo, generation) do`
	295	+	`table = ensure_index_cache_table()`
	296	+
	297	+	`case :ets.lookup(table, repo) do`
	298	+	`[{^repo, ^generation, index}] -> {:ok, index}`
	299	+	`_other -> :miss`
	300	+	`end`
	301	+	`end`
	302	+
	303	+	`defp cache_index(repo, generation, index) do`
	304	+	`:ets.insert(ensure_index_cache_table(), {repo, generation, index})`
	305	+	`:ok`
	306	+	`end`
	307	+
	308	+	`defp delete_cached_index(repo) do`
	309	+	`case :ets.whereis(@index_cache_table) do`
	310	+	`:undefined -> :ok`
	311	+	`table -> :ets.delete(table, repo)`
	312	+	`end`
	313	+	`end`
	314	+
	315	+	`defp ensure_index_cache_table do`
	316	+	`case :ets.whereis(@index_cache_table) do`
	317	+	`:undefined ->`
	318	+	`heir = Process.whereis(OpenAgents.Forge.CacheReadiness)`
	319	+
	320	+	`options =`
	321	+	`[:named_table, :public, :set, read_concurrency: true] ++`
	322	+	`if(is_pid(heir), do: [{:heir, heir, :wal_index_cache}], else: [])`
	323	+
	324	+	`try do`
	325	+	`:ets.new(@index_cache_table, options)`
	326	+	`rescue`
	327	+	`ArgumentError -> @index_cache_table`
	328	+	`end`
	329	+
	330	+	`table ->`
	331	+	`table`
	332	+	`end`
	333	+	`end`
	334	+
	335	+	`defp request_options(options) do`
	336	+	`Keyword.merge(`
	337	+	`options,`
	338	+	`Application.get_env(:openagents, :forge_gcs_request_options, [])`
	339	+	`)`
	340	+	`end`
	341	+
264	342		`defp upload_file(bucket, name, path, size, token) do`
265	343		`url =`
266	344		`@storage_base <>`

      raise OpenAgentsWeb.PublicNotFoundError
    end

    sha =
      case Browse.resolve_commit(repository, ref) do
        {:ok, sha} -> sha
    %{sha: sha, head: head, blob: blob} =
      case Browse.blob_page(repository, ref, path) do
        {:ok, page} -> page
        _ -> raise OpenAgentsWeb.PublicNotFoundError
      end

    # Either the whole repo is browsable (:l3), or this is a published
    # document at the current head. A published path at an older ref is a
    # 404: publishing one document must not publish its history.
    head = with {:ok, head} <- Browse.head(repository), do: head

    unless RepositoryAccess.allows_file?(
             repository,
             socket.assigns.current_user,

      raise OpenAgentsWeb.PublicNotFoundError
    end

    blob =
      case Browse.blob(repository, sha, path) do
        {:ok, blob} -> blob
        _ -> raise OpenAgentsWeb.PublicNotFoundError
      end

    plain = Map.get(params, "plain") == "1"

    {:ok,

    end
  end

  test "blob_page/3 returns a resolved revision, head, and blob together", %{second: second} do
    assert {:ok, page} = Browse.blob_page("openagents.com", "main", "file.txt")

    assert page.sha == second
    assert page.head == second
    assert page.blob.content == "hello world\n"
  end

  test "readme/2 finds README.md at a ref" do
    assert {:ok, "README.md", blob} = Browse.readme("openagents.com", "main")
    assert blob.content =~ "Fixture readme."

  use ExUnit.Case, async: false
  alias OpenAgents.Forge.WAL

  setup {Req.Test, :verify_on_exit!}

  @repo "openagents.com"

  setup do

  end

  describe "GCS adapter (offline)" do
    setup do
      OpenAgents.Forge.WAL.Gcs.reset_index_cache()
      :ok
    end

    test "returns :not_configured when no bucket is set" do
      previous = Application.fetch_env(:openagents, :forge_wal_bucket)
      Application.delete_env(:openagents, :forge_wal_bucket)

418	425		`)`
419	426		`end`
420	427
	428	+	`test "reuses a decoded index while the GCS generation is unchanged" do`
	429	+	`previous_bucket = Application.fetch_env(:openagents, :forge_wal_bucket)`
	430	+	`previous_provider = Application.fetch_env(:openagents, :forge_gcs_token_provider)`
	431	+	`previous_options = Application.fetch_env(:openagents, :forge_gcs_request_options)`
	432	+
	433	+	`Application.put_env(:openagents, :forge_wal_bucket, "test-bucket")`
	434	+	`Application.put_env(:openagents, :forge_gcs_token_provider, fn -> "test-token" end)`
	435	+
	436	+	`Application.put_env(:openagents, :forge_gcs_request_options, plug: {Req.Test, __MODULE__})`
	437	+
	438	+	`on_exit(fn ->`
	439	+	`restore_env(:forge_wal_bucket, previous_bucket)`
	440	+	`restore_env(:forge_gcs_token_provider, previous_provider)`
	441	+	`restore_env(:forge_gcs_request_options, previous_options)`
	442	+	`end)`
	443	+
	444	+	`index = WAL.new_index()`
	445	+
	446	+	`Req.Test.expect(__MODULE__, fn request ->`
	447	+	`assert request.query_string == "fields=generation"`
	448	+	`Req.Test.json(request, %{"generation" => "42"})`
	449	+	`end)`
	450	+
	451	+	`Req.Test.expect(__MODULE__, fn request ->`
	452	+	`assert request.query_string == "alt=media"`
	453	+	`Req.Test.json(request, index)`
	454	+	`end)`
	455	+
	456	+	`Req.Test.expect(__MODULE__, fn request ->`
	457	+	`assert request.query_string == "fields=generation"`
	458	+	`Req.Test.json(request, %{"generation" => "42"})`
	459	+	`end)`
	460	+
	461	+	`assert {:ok, "42", ^index} = OpenAgents.Forge.WAL.Gcs.read_index(@repo)`
	462	+	`assert {:ok, "42", ^index} = OpenAgents.Forge.WAL.Gcs.read_index(@repo)`
	463	+	`end`
	464	+
421	465		`test "object naming helpers" do`
422	466		`assert OpenAgents.Forge.WAL.Gcs.prefix("openagents.com") == "forge/wal/openagents.com/"`
423	467

273	273		`end`
274	274		`end`
275	275
	276	+	`@doc "Read the data for a file page after one cache-freshness check."`
	277	+	`def blob_page(repo, ref, path) do`
	278	+	`with :ok <- check(repo, ref),`
	279	+	`:ok <- check_path(path) do`
	280	+	`_ = freshen(repo)`
	281	+
	282	+	`with {:ok, sha} <- resolve_commit_from_cache(repo, ref),`
	283	+	`{:ok, blob} <- blob_from_cache(repo, sha, path) do`
	284	+	`head = with {:ok, current} <- head_from_cache(repo), do: current`
	285	+	`{:ok, %{sha: sha, head: head, blob: blob}}`
	286	+	`end`
	287	+	`end`
	288	+	`end`
	289	+
276	290		`defp blob_from_cache(repo, ref, path) do`
277	291		`with {:ok, full} <- resolve_commit_from_cache(repo, ref) do`
278	292		`spec = full <> ":" <> path`

333	333		`end`
334	334
335	335		`defp refs_materialized_at?(path, index) do`
336		-	`index`
337		-	`\|> WAL.refs()`
338		-	`\|> Map.values()`
339		-	`\|> Enum.uniq()`
340		-	`\|> Enum.all?(fn sha ->`
341		-	`match?({_output, 0}, Repos.git(path, ["cat-file", "-e", sha]))`
342		-	`end)`
	336	+	`tips = index \|> WAL.refs() \|> Map.values() \|> Enum.uniq()`
	337	+
	338	+	`case tips do`
	339	+	`[] ->`
	340	+	`true`
	341	+
	342	+	`tips ->`
	343	+	`input = Enum.join(tips, "\n") <> "\n"`
	344	+
	345	+	`case Repos.git_with_stdin(path, ["cat-file", "--batch-check"], input) do`
	346	+	`{output, 0} ->`
	347	+	`lines = String.split(output, "\n", trim: true)`
	348	+
	349	+	`length(lines) == length(tips) and`
	350	+	`Enum.all?(lines, &(not String.ends_with?(&1, " missing")))`
	351	+
	352	+	`_error ->`
	353	+	`false`
	354	+	`end`
	355	+	`end`
343	356		`end`
344	357
345	358		`# Apply one WAL entry: materialize its objects, prove the objects it`

63	72		`headers: auth_headers(token) ++ [{"content-type", "application/json"}]`
64	73		`) do`
65	74		`{:ok, %Req.Response{status: 200, body: %{"generation" => generation}}} ->`
	75	+	`cache_index(repo, generation, index)`
66	76		`{:ok, generation}`
67	77
68	78		`{:ok, %Req.Response{status: 412}} ->`

149	159		`def delete_repo(repo) do`
150	160		`with {:ok, bucket} <- bucket(),`
151	161		`{:ok, token} <- token() do`
152		-	`delete_prefix(bucket, prefix(repo), token)`
	162	+	`case delete_prefix(bucket, prefix(repo), token) do`
	163	+	`:ok ->`
	164	+	`delete_cached_index(repo)`
	165	+	`:ok`
	166	+
	167	+	`{:error, _reason} = error ->`
	168	+	`error`
	169	+	`end`
153	170		`end`
154	171		`end`
155	172
	173	+	`@doc false`
	174	+	`def reset_index_cache do`
	175	+	`case :ets.whereis(@index_cache_table) do`
	176	+	`:undefined -> :ok`
	177	+	`_table -> :ets.delete_all_objects(@index_cache_table)`
	178	+	`end`
	179	+
	180	+	`:ok`
	181	+	`end`
	182	+
156	183		`## Object naming (public so it is testable without a live bucket)`
157	184
158	185		`@doc """`

232	259		`with {:ok, token} <- token() do`
233	260		`url = object_url(bucket, name) <> "?" <> URI.encode_query(fields: "generation")`
234	261
235		-	`case Req.get(url, headers: auth_headers(token)) do`
	262	+	`case Req.get(url, request_options(headers: auth_headers(token))) do`
236	263		`{:ok, %Req.Response{status: 200, body: %{"generation" => generation}}} ->`
237	264		`{:ok, generation}`
238	265

Commit

Speed up forge blob pages

Deploy story

Changed files

Diff

25	25		`@storage_base "https://storage.googleapis.com"`
26	26		`@metadata_token_url "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token"`
27	27		`@token_cache_key {__MODULE__, :token}`
	28	+	`@index_cache_table OpenAgents.Forge.WAL.Gcs.IndexCache`
28	29		`@token_expiry_margin_seconds 60`
29	30		`@stream_chunk_bytes 1_048_576`
30	31		`@default_stream_timeout_ms 6 * 60 * 60 * 1_000`

34	35		`with {:ok, bucket} <- bucket() do`
35	36		`name = index_object(repo)`
36	37
37		-	`with {:ok, generation} <- fetch_generation(bucket, name),`
38		-	`{:ok, raw} <- download(bucket, name),`
39		-	`{:ok, index} <- decode_index(raw) do`
40		-	`{:ok, generation, index}`
	38	+	`with {:ok, generation} <- fetch_generation(bucket, name) do`
	39	+	`case cached_index(repo, generation) do`
	40	+	`{:ok, index} ->`
	41	+	`{:ok, generation, index}`
	42	+
	43	+	`:miss ->`
	44	+	`with {:ok, raw} <- download(bucket, name),`
	45	+	`{:ok, index} <- decode_index(raw) do`
	46	+	`cache_index(repo, generation, index)`
	47	+	`{:ok, generation, index}`
	48	+	`end`
	49	+	`end`
41	50		`end`
42	51		`end`
43	52		`end`

252	279		`with {:ok, token} <- token() do`
253	280		`url = object_url(bucket, name) <> "?alt=media"`
254	281
255		-	`case Req.get(url, headers: auth_headers(token), decode_body: false) do`
	282	+	`case Req.get(`
	283	+	`url,`
	284	+	`request_options(headers: auth_headers(token), decode_body: false)`
	285	+	`) do`
256	286		`{:ok, %Req.Response{status: 200, body: body}} -> {:ok, body}`
257	287		`{:ok, %Req.Response{status: 404}} -> {:error, :not_found}`
258	288		`{:ok, %Req.Response{status: status, body: body}} -> {:error, {:gcs_error, status, body}}`

29	29		`raise OpenAgentsWeb.PublicNotFoundError`
30	30		`end`
31	31
32		-	`sha =`
33		-	`case Browse.resolve_commit(repository, ref) do`
34		-	`{:ok, sha} -> sha`
	32	+	`%{sha: sha, head: head, blob: blob} =`
	33	+	`case Browse.blob_page(repository, ref, path) do`
	34	+	`{:ok, page} -> page`
35	35		`_ -> raise OpenAgentsWeb.PublicNotFoundError`
36	36		`end`
37	37
38	38		`# Either the whole repo is browsable (:l3), or this is a published`
39	39		`# document at the current head. A published path at an older ref is a`
40	40		`# 404: publishing one document must not publish its history.`
41		-	`head = with {:ok, head} <- Browse.head(repository), do: head`
42		-
43	41		`unless RepositoryAccess.allows_file?(`
44	42		`repository,`
45	43		`socket.assigns.current_user,`

50	48		`raise OpenAgentsWeb.PublicNotFoundError`
51	49		`end`
52	50
53		-	`blob =`
54		-	`case Browse.blob(repository, sha, path) do`
55		-	`{:ok, blob} -> blob`
56		-	`_ -> raise OpenAgentsWeb.PublicNotFoundError`
57		-	`end`
58		-
59	51		`plain = Map.get(params, "plain") == "1"`
60	52
61	53		`{:ok,`

201	201		`end`
202	202		`end`
203	203
	204	+	`test "blob_page/3 returns a resolved revision, head, and blob together", %{second: second} do`
	205	+	`assert {:ok, page} = Browse.blob_page("openagents.com", "main", "file.txt")`
	206	+
	207	+	`assert page.sha == second`
	208	+	`assert page.head == second`
	209	+	`assert page.blob.content == "hello world\n"`
	210	+	`end`
	211	+
204	212		`test "readme/2 finds README.md at a ref" do`
205	213		`assert {:ok, "README.md", blob} = Browse.readme("openagents.com", "main")`
206	214		`assert blob.content =~ "Fixture readme."`

2	2		`use ExUnit.Case, async: false`
3	3		`alias OpenAgents.Forge.WAL`
4	4
	5	+	`setup {Req.Test, :verify_on_exit!}`
	6	+
5	7		`@repo "openagents.com"`
6	8
7	9		`setup do`

391	393		`end`
392	394
393	395		`describe "GCS adapter (offline)" do`
	396	+	`setup do`
	397	+	`OpenAgents.Forge.WAL.Gcs.reset_index_cache()`
	398	+	`:ok`
	399	+	`end`
	400	+
394	401		`test "returns :not_configured when no bucket is set" do`
395	402		`previous = Application.fetch_env(:openagents, :forge_wal_bucket)`
396	403		`Application.delete_env(:openagents, :forge_wal_bucket)`