move protobuf definitions to become dependency of both client and server

Author: InvalidPathException

.formatter.exs

@@ -4,7 +4,6 @@
   plugins: [Phoenix.LiveView.HTMLFormatter],
   inputs: [
     "*.{heex,ex,exs}",
-    "{config,lib,test}/**/*.{heex,ex,exs}",
-    "apps/*/lib/**/*.{heex,ex,exs}"
+    "{config,lib,test}/**/*.{heex,ex,exs}"
   ]
 ]

apps/kafkaesque_client/mix.exs

@@ -28,8 +28,7 @@ defmodule KafkaesqueClient.MixProject do
 
   defp deps do
     [
-      {:grpc, "~> 0.7"},
-      {:protobuf, "~> 0.12"},
+      {:kafkaesque_proto, in_umbrella: true},
       {:gun, "~> 2.0"},
       {:jason, "~> 1.4"},
       {:telemetry, "~> 1.2"},

apps/kafkaesque_core/lib/kafkaesque/storage/single_file.ex

@@ -82,6 +82,10 @@ defmodule Kafkaesque.Storage.SingleFile do
     GenServer.call(via_tuple(topic, partition), :close)
   end
 
+  def get_index_table(topic, partition) do
+    GenServer.call(via_tuple(topic, partition), :get_index_table)
+  end
+
   @impl true
   def init(opts) do
     topic = Keyword.fetch!(opts, :topic)
@@ -240,6 +244,11 @@ defmodule Kafkaesque.Storage.SingleFile do
     {:reply, {:ok, offsets}, state}
   end
 
+  @impl true
+  def handle_call(:get_index_table, _from, state) do
+    {:reply, {:ok, state.index.table}, state}
+  end
+
   @impl true
   def handle_call(:close, _from, state) do
     # Flush and close file

apps/kafkaesque_core/lib/kafkaesque/topic/retention_controller.ex

@@ -202,7 +202,6 @@ defmodule Kafkaesque.Topic.RetentionController do
       cutoff_time = current_time - state.retention_ms
 
       # Find the offset of the first message after the cutoff time
-      # In a real implementation, this would scan the index
       find_offset_after_timestamp(state, cutoff_time)
     end
   end
@@ -212,7 +211,6 @@ defmodule Kafkaesque.Topic.RetentionController do
       0
     else
       # Calculate total size and find offset where we exceed retention
-      # In a real implementation, this would use the index and file stats
       find_offset_for_size_limit(state, state.retention_bytes)
     end
   end
@@ -231,8 +229,19 @@ defmodule Kafkaesque.Topic.RetentionController do
     # Get offsets range first
     case SingleFile.get_offsets(state.topic, state.partition) do
       {:ok, %{earliest: earliest, latest: latest}} when latest > earliest ->
-        # Use binary search to find the first offset after cutoff time
-        binary_search_timestamp(state.topic, state.partition, earliest, latest, cutoff_time)
+        # Try to get index table for more efficient search
+        index_table = get_index_table(state)
+
+        # Use index to narrow search if available
+        {search_start, search_end} =
+          if index_table do
+            narrow_search_with_index(index_table, earliest, latest, cutoff_time)
+          else
+            {earliest, latest}
+          end
+
+        # Binary search to find the first offset after cutoff time
+        binary_search_timestamp(state.topic, state.partition, search_start, search_end, cutoff_time)
 
       _ ->
         # No messages or error
@@ -290,19 +299,27 @@ defmodule Kafkaesque.Topic.RetentionController do
         # We need to keep only size_limit_bytes, so calculate how much to remove
         bytes_to_remove = file_size - size_limit_bytes
 
+        # Try to use index for more efficient size calculation
+        index_table = get_index_table(state)
+
         # Scan through messages to find where we've accumulated enough bytes to remove
-        find_size_based_offset(state.topic, state.partition, bytes_to_remove)
+        find_size_based_offset(state.topic, state.partition, bytes_to_remove, index_table)
 
       _ ->
         # File is smaller than limit, keep everything
         0
     end
   end
 
-  defp find_size_based_offset(topic, partition, bytes_to_remove) do
+  defp find_size_based_offset(topic, partition, bytes_to_remove, index_table) do
     case SingleFile.get_offsets(topic, partition) do
       {:ok, %{earliest: earliest, latest: latest}} when latest > earliest ->
-        scan_for_size_limit_recursive(topic, partition, earliest, latest, bytes_to_remove, 0)
+        # Use index entries to jump more efficiently if available
+        if index_table do
+          scan_with_index(topic, partition, index_table, bytes_to_remove, earliest, latest)
+        else
+          scan_for_size_limit_recursive(topic, partition, earliest, latest, bytes_to_remove, 0)
+        end
 
       _ ->
         0
@@ -475,4 +492,85 @@ defmodule Kafkaesque.Topic.RetentionController do
       newest_timestamp: System.system_time(:millisecond)
     })
   end
+
+  defp get_index_table(state) do
+    # Get the index table from SingleFile
+    case SingleFile.get_index_table(state.topic, state.partition) do
+      {:ok, table} -> table
+      _ -> nil
+    end
+  rescue
+    _ -> nil
+  end
+
+  defp narrow_search_with_index(index_table, earliest, latest, _cutoff_time) do
+    # Use index entries to narrow binary search range
+    # Since we don't have timestamps in index yet, we can at least use
+    # the index to skip large ranges
+    entries = :ets.tab2list(index_table) |> Enum.sort()
+
+    # For now, just use index to segment the search space
+    # TODO: We could later enhance index to store timestamps
+    if length(entries) > 10 do
+      # Use middle third of the range for search (heuristic)
+      range = latest - earliest
+      search_start = earliest + div(range, 3)
+      search_end = latest - div(range, 10)
+      {search_start, search_end}
+    else
+      {earliest, latest}
+    end
+  end
+
+  defp scan_with_index(topic, partition, index_table, bytes_to_remove, earliest, latest) do
+    # Use index entries to jump through the file more efficiently
+    entries = :ets.tab2list(index_table) |> Enum.sort()
+
+    # Calculate approximate bytes per offset using index positions
+    if length(entries) > 1 do
+      # Use index to estimate where size limit is reached
+      scan_with_index_entries(topic, partition, entries, bytes_to_remove, 0)
+    else
+      # Fall back to regular scan
+      scan_for_size_limit_recursive(topic, partition, earliest, latest, bytes_to_remove, 0)
+    end
+  end
+
+  defp scan_with_index_entries(topic, partition, entries, bytes_to_remove, accumulated) do
+    case entries do
+      [{offset, {position, _frame_len}} | rest] when rest != [] ->
+        [{next_offset, {next_position, _}} | _] = rest
+
+        # Approximate size between these index entries
+        segment_size = next_position - position
+        new_accumulated = accumulated + segment_size
+
+        if new_accumulated >= bytes_to_remove do
+          # Found the region, scan this segment in detail
+          scan_for_exact_offset_in_segment(topic, partition, offset, next_offset,
+                                          bytes_to_remove, accumulated)
+        else
+          scan_with_index_entries(topic, partition, rest, bytes_to_remove, new_accumulated)
+        end
+
+      _ ->
+        # End of index or single entry
+        0
+    end
+  end
+
+  defp scan_for_exact_offset_in_segment(topic, partition, start_offset, end_offset,
+                                       bytes_to_remove, accumulated_before) do
+    # Scan the specific segment to find exact offset
+    batch_size = min(50, end_offset - start_offset + 1)
+
+    case SingleFile.read(topic, partition, start_offset, batch_size * 100) do
+      {:ok, messages} when messages != [] ->
+        # Find exact cutoff in this batch
+        find_exact_size_cutoff(messages, start_offset, bytes_to_remove, accumulated_before)
+
+      _ ->
+        start_offset
+    end
+  end
 end

apps/kafkaesque_core/test/test_helper.exs

@@ -3,7 +3,7 @@ Application.put_env(:kafkaesque_core, :data_dir, "/tmp/kafkaesque_test/data")
 Application.put_env(:kafkaesque_core, :offsets_dir, "/tmp/kafkaesque_test/offsets")
 
 # Configure faster batching for tests
-# NOTE: All timeout values are in milliseconds for consistency
+# All timeout values are in milliseconds
 Application.put_env(:kafkaesque_core, :batch_size, 5)  # Smaller batches
 Application.put_env(:kafkaesque_core, :batch_timeout, 100)  # 100ms for faster tests
 Application.put_env(:kafkaesque_core, :fsync_interval_ms, 50)  # Faster fsync

apps/kafkaesque_proto/.formatter.exs

@@ -0,0 +1,9 @@
+# Used by "mix format"
+[
+  inputs: [
+    "{mix,.formatter}.exs",
+    "{config,test}/**/*.{ex,exs}",
+    "lib/kafkaesque_proto.ex"
+    # lib/kafkaesque.pb.ex is excluded as it's auto-generated
+  ]
+]

apps/kafkaesque_proto/.gitignore

@@ -0,0 +1,23 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+kafkaesque_proto-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

apps/kafkaesque_proto/README.md

@@ -0,0 +1,81 @@
+# KafkaesqueProto
+
+Shared protobuf definitions for the Kafkaesque distributed log system.
+
+## Purpose
+
+This umbrella app contains the protobuf definitions that are shared between:
+- `kafkaesque_server` - The gRPC server implementation
+- `kafkaesque_client` - The gRPC client SDK
+
+By centralizing the protobuf definitions in a single app, we ensure:
+- Single source of truth for API contracts
+- Automatic synchronization between client and server
+- No code duplication or version drift
+
+## Structure
+
+```
+apps/kafkaesque_proto/
+├── lib/
+│   └── kafkaesque.pb.ex       # Generated protobuf Elixir code
+├── priv/
+│   └── protos/
+│       └── kafkaesque.proto   # Source protobuf definitions
+└── mix.exs
+```
+
+## Usage
+
+### Generating Protobuf Code
+
+From the project root:
+```bash
+mix proto.gen
+```
+
+Or from this app directory with PATH set correctly:
+```bash
+export PATH=$PATH:~/.mix/escripts
+mix proto.gen
+```
+
+### Using in Other Apps
+
+Add to your `mix.exs` dependencies:
+```elixir
+{:kafkaesque_proto, in_umbrella: true}
+```
+
+Then use the generated modules:
+```elixir
+alias Kafkaesque.{ProduceRequest, ProduceResponse}
+alias Kafkaesque.Kafkaesque.Stub  # For client
+alias Kafkaesque.Kafkaesque.Service  # For server
+```
+
+## API Messages
+
+The protobuf definitions include:
+- **Topic Management**: CreateTopicRequest, ListTopicsRequest/Response
+- **Producing**: ProduceRequest/Response with Acks enum
+- **Consuming**: ConsumeRequest, FetchedBatch (streaming)
+- **Offsets**: CommitOffsetsRequest/Response, GetOffsetsRequest/Response
+- **Core Types**: Topic, Record, RecordHeader
+
+## Development
+
+When modifying the proto file:
+1. Edit `priv/protos/kafkaesque.proto`
+2. Run `mix proto.gen` to regenerate code
+3. Recompile dependent apps with `mix compile --force`
+
+## Prerequisites
+
+Ensure you have:
+- `protoc` (Protocol Buffers compiler) installed
+- `protoc-gen-elixir` plugin installed:
+  ```bash
+  mix escript.install hex protobuf
+  export PATH=$PATH:~/.mix/escripts
+  ```

…kaesque_server/lib/grpc/kafkaesque.pb.ex …ps/kafkaesque_proto/lib/kafkaesque.pb.exapps/kafkaesque_server/lib/grpc/kafkaesque.pb.ex renamed to apps/kafkaesque_proto/lib/kafkaesque.pb.ex apps/kafkaesque_server/lib/grpc/kafkaesque.pb.ex renamed to apps/kafkaesque_proto/lib/kafkaesque.pb.ex

@@ -137,6 +137,12 @@ defmodule Kafkaesque.CreateTopicRequest do
   field(:partitions, 2, type: :int32)
 end
 
+defmodule Kafkaesque.ListTopicsRequest do
+  @moduledoc false
+
+  use Protobuf, protoc_gen_elixir_version: "0.15.0", syntax: :proto3
+end
+
 defmodule Kafkaesque.ListTopicsResponse do
   @moduledoc false
 
@@ -152,7 +158,7 @@ defmodule Kafkaesque.Kafkaesque.Service do
 
   rpc(:CreateTopic, Kafkaesque.CreateTopicRequest, Kafkaesque.Topic)
 
-  rpc(:ListTopics, Google.Protobuf.Empty, Kafkaesque.ListTopicsResponse)
+  rpc(:ListTopics, Kafkaesque.ListTopicsRequest, Kafkaesque.ListTopicsResponse)
 
   rpc(:Produce, Kafkaesque.ProduceRequest, Kafkaesque.ProduceResponse)
 

apps/kafkaesque_proto/lib/kafkaesque_proto.ex

@@ -0,0 +1,18 @@
+defmodule KafkaesqueProto do
+  @moduledoc """
+  Documentation for `KafkaesqueProto`.
+  """
+
+  @doc """
+  Hello world.
+
+  ## Examples
+
+      iex> KafkaesqueProto.hello()
+      :world
+
+  """
+  def hello do
+    :world
+  end
+end