nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: benbjohnson/litestream
Pull request #345 (2022-03-05 18:30:03)

Prevent double-close for SFTP client
Typically, executing `Close()` multiple times is ok as duplicates are typically treated as a no-op. However, the SFTP client's file handles will send duplicate `CLOSE` messages if called twice. This pull request adds a `OnceCloser` implementation to ensure `Close()` is only executed once.

Fixes https://github.com/benbjohnson/litestream/issues/268


Patch:
diff --git a/internal/internal.go b/internal/internal.go
index c671379..95d0f78 100644
--- a/internal/internal.go
+++ b/internal/internal.go
@@ -8,6 +8,7 @@ import (
 	"os"
 	"regexp"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -264,3 +265,18 @@ func TruncateDuration(d time.Duration) time.Duration {
 func MD5Hash(b []byte) string {
 	return fmt.Sprintf("%x", md5.Sum(b))
 }
+
+// OnceCloser returns a closer that will only ignore duplicate closes.
+func OnceCloser(c io.Closer) io.Closer {
+	return &onceCloser{Closer: c}
+}
+
+type onceCloser struct {
+	sync.Once
+	io.Closer
+}
+
+func (c *onceCloser) Close() (err error) {
+	c.Once.Do(func() { err = c.Closer.Close() })
+	return err
+}
diff --git a/sftp/replica_client.go b/sftp/replica_client.go
index 8566ab0..269e8ca 100644
--- a/sftp/replica_client.go
+++ b/sftp/replica_client.go
@@ -270,12 +270,13 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 	if err != nil {
 		return info, fmt.Errorf("cannot open snapshot file for writing: %w", err)
 	}
-	defer f.Close()
+	closer := internal.OnceCloser(f)
+	defer closer.Close()
 
 	n, err := io.Copy(f, rd)
 	if err != nil {
 		return info, err
-	} else if err := f.Close(); err != nil {
+	} else if err := closer.Close(); err != nil {
 		return info, err
 	}
 
@@ -391,12 +392,13 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
 	if err != nil {
 		return info, fmt.Errorf("cannot open snapshot file for writing: %w", err)
 	}
-	defer f.Close()
+	closer := internal.OnceCloser(f)
+	defer closer.Close()
 
 	n, err := io.Copy(f, rd)
 	if err != nil {
 		return info, err
-	} else if err := f.Close(); err != nil {
+	} else if err := closer.Close(); err != nil {
 		return info, err
 	}
 


Test patch:
diff --git a/internal/internal_test.go b/internal/internal_test.go
index 9d2c49b..11d0f6d 100644
--- a/internal/internal_test.go
+++ b/internal/internal_test.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/benbjohnson/litestream/internal"
+	"github.com/benbjohnson/litestream/mock"
 )
 
 func TestParseSnapshotPath(t *testing.T) {
@@ -100,3 +101,27 @@ func TestTruncateDuration(t *testing.T) {
 		})
 	}
 }
+
+func TestOnceCloser(t *testing.T) {
+	var closed bool
+	var rc = &mock.ReadCloser{
+		CloseFunc: func() error {
+			if closed {
+				t.Fatal("already closed")
+			}
+			closed = true
+			return nil
+		},
+	}
+
+	oc := internal.OnceCloser(rc)
+	if err := oc.Close(); err != nil {
+		t.Fatalf("first close: %s", err)
+	} else if err := oc.Close(); err != nil {
+		t.Fatalf("second close: %s", err)
+	}
+
+	if !closed {
+		t.Fatal("expected close")
+	}
+}
Repository: benbjohnson/litestream
Pull request #620 (2024-12-09 22:23:03)

Add busy-timeout-interval setting
Adds a `busy-timeout` yaml setting that adjusts the `PRAGMA busy_timeout` in litestream to handle applications that hold the lock for longer than the default value of 1 second. 

(i.e. the converse of setting the https://litestream.io/tips/#busy-timeout in the application) 

darkgnotic: > Can you change the name from "busy timeout interval" to just "busy timeout"? It's not really an interval since it's not repeating. Otherwise the PR looks good.

Ah, good point! Done!


Patch:
diff --git a/db.go b/db.go
index 85256a2..debf5a5 100644
--- a/db.go
+++ b/db.go
@@ -28,6 +28,7 @@ import (
 const (
 	DefaultMonitorInterval    = 1 * time.Second
 	DefaultCheckpointInterval = 1 * time.Minute
+	DefaultBusyTimeout        = 1 * time.Second
 	DefaultMinCheckpointPageN = 1000
 	DefaultMaxCheckpointPageN = 10000
 	DefaultTruncatePageN      = 500000
@@ -37,9 +38,6 @@ const (
 // If this index is reached then a new generation will be started.
 const MaxIndex = 0x7FFFFFFF
 
-// BusyTimeout is the timeout to wait for EBUSY from SQLite.
-const BusyTimeout = 1 * time.Second
-
 // DB represents a managed instance of a SQLite database in the file system.
 type DB struct {
 	mu       sync.RWMutex
@@ -103,6 +101,9 @@ type DB struct {
 	// Frequency at which to perform db sync.
 	MonitorInterval time.Duration
 
+	// The timeout to wait for EBUSY from SQLite.
+	BusyTimeout time.Duration
+
 	// List of replicas for the database.
 	// Must be set before calling Open().
 	Replicas []*Replica
@@ -125,6 +126,7 @@ func NewDB(path string) *DB {
 		TruncatePageN:      DefaultTruncatePageN,
 		CheckpointInterval: DefaultCheckpointInterval,
 		MonitorInterval:    DefaultMonitorInterval,
+		BusyTimeout:        DefaultBusyTimeout,
 		Logger:             slog.With("db", path),
 	}
 
@@ -411,7 +413,7 @@ func (db *DB) init() (err error) {
 	db.dirInfo = fi
 
 	dsn := db.path
-	dsn += fmt.Sprintf("?_busy_timeout=%d", BusyTimeout.Milliseconds())
+	dsn += fmt.Sprintf("?_busy_timeout=%d", db.BusyTimeout.Milliseconds())
 
 	// Connect to SQLite database. Use the driver registered with a hook to
 	// prevent WAL files from being removed.


Test patch:
diff --git a/cmd/litestream/main.go b/cmd/litestream/main.go
index 9698bf5..0115e55 100644
--- a/cmd/litestream/main.go
+++ b/cmd/litestream/main.go
@@ -287,6 +287,7 @@ type DBConfig struct {
 	MetaPath           *string        `yaml:"meta-path"`
 	MonitorInterval    *time.Duration `yaml:"monitor-interval"`
 	CheckpointInterval *time.Duration `yaml:"checkpoint-interval"`
+	BusyTimeout        *time.Duration `yaml:"busy-timeout"`
 	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
 	MaxCheckpointPageN *int           `yaml:"max-checkpoint-page-count"`
 
@@ -313,6 +314,9 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	if dbc.CheckpointInterval != nil {
 		db.CheckpointInterval = *dbc.CheckpointInterval
 	}
+	if dbc.BusyTimeout != nil {
+		db.BusyTimeout = *dbc.BusyTimeout
+	}
 	if dbc.MinCheckpointPageN != nil {
 		db.MinCheckpointPageN = *dbc.MinCheckpointPageN
 	}
Repository: livebook-dev/kino
Pull request #271 (2023-03-17 17:21:21)

Do not terminate async_listen processes when stream finishes
🐈


Patch:
diff --git a/lib/kino.ex b/lib/kino.ex
index 2772881..199423d 100644
--- a/lib/kino.ex
+++ b/lib/kino.ex
@@ -470,6 +470,16 @@ defmodule Kino do
       end
 
       after_fun = fn task_supervisor ->
+        for {_, pid, _, _} <- DynamicSupervisor.which_children(task_supervisor),
+            is_pid(pid),
+            Process.alive?(pid) do
+          ref = Process.monitor(pid)
+
+          receive do
+            {:DOWN, ^ref, _, _, _} -> :ok
+          end
+        end
+
         DynamicSupervisor.terminate_child(Kino.DynamicSupervisor, task_supervisor)
       end
 


Test patch:
diff --git a/test/kino_test.exs b/test/kino_test.exs
index 5c8ea87..afb287d 100644
--- a/test/kino_test.exs
+++ b/test/kino_test.exs
@@ -266,6 +266,25 @@ defmodule KinoTest do
       assert log =~ "Kino.async_listen"
       assert log =~ "** (MatchError) no match of right hand side value: 0"
     end
+
+    test "processes keep running when the stream finishes" do
+      myself = self()
+
+      Stream.iterate(0, &(&1 + 1))
+      |> Stream.take(2)
+      |> Kino.async_listen(fn _i ->
+        send(myself, {:item, self()})
+        Process.sleep(:infinity)
+      end)
+
+      assert_receive {:item, pid1}
+      assert_receive {:item, pid2}
+
+      Process.sleep(1)
+
+      assert Process.alive?(pid1)
+      assert Process.alive?(pid2)
+    end
   end
 
   defp await_process(pid) do
Repository: livebook-dev/kino
Pull request #369 (2023-11-23 15:14:53)

Don't discard currently registered dbg_callback
Fixes https://github.com/elixir-lang/elixir/issues/12687#issuecomment-1820525658.

jonatanklosko: We have a test, so it needs to be changed too :)

Patch:
diff --git a/lib/kino/application.ex b/lib/kino/application.ex
index d154830..ef6b77e 100644
--- a/lib/kino/application.ex
+++ b/lib/kino/application.ex
@@ -4,7 +4,8 @@ defmodule Kino.Application do
   use Application
 
   def start(_type, _args) do
-    Application.put_env(:elixir, :dbg_callback, {Kino.Debug, :dbg, []})
+    original = Application.fetch_env!(:elixir, :dbg_callback)
+    Application.put_env(:elixir, :dbg_callback, {Kino.Debug, :dbg, [original]})
 
     Kino.AttributeStore.initialize()
 
diff --git a/lib/kino/debug.ex b/lib/kino/debug.ex
index 77b2d52..94030f8 100644
--- a/lib/kino/debug.ex
+++ b/lib/kino/debug.ex
@@ -10,8 +10,9 @@ defmodule Kino.Debug do
   `Kernel.dbg/2` calls in certain cases, such as call pipelines. It
   falls back to the default backend otherwise.
   """
-  @spec dbg(Macro.t(), Macro.t(), Macro.Env.t()) :: Macro.t()
-  def dbg(ast, options, %Macro.Env{} = env) do
+  @spec dbg(Macro.t(), Macro.t(), Macro.Env.t(), tuple()) :: Macro.t()
+  def dbg(ast, options, %Macro.Env{} = env, previous_dbg) do
+    {previous_mod, previous_fun, previous_args} = previous_dbg
     dbg_id = System.unique_integer()
 
     kino_ast =
@@ -23,7 +24,7 @@ defmodule Kino.Debug do
           dbg_default_ast(ast, dbg_id, env)
       end
 
-    fallback_ast = Macro.dbg(ast, options, env)
+    fallback_ast = apply(previous_mod, previous_fun, [ast, options, env | previous_args])
 
     quote do
       if Kino.Bridge.within_livebook?() do


Test patch:
diff --git a/test/kino/debug_test.exs b/test/kino/debug_test.exs
index bebd7e7..7eff841 100644
--- a/test/kino/debug_test.exs
+++ b/test/kino/debug_test.exs
@@ -3,7 +3,7 @@ defmodule Kino.Debug.Test do
 
   defmacrop call_dbg(ast) do
     quote do
-      unquote(Kino.Debug.dbg(ast, [], __CALLER__))
+      unquote(Kino.Debug.dbg(ast, [], __CALLER__, {Macro, :dbg, []}))
     end
   end
Repository: livebook-dev/kino
Pull request #485 (2024-12-25 16:48:05)

Allow any term to be used as tag in Kino.Control.tagged_stream/1
Supersedes #484.

Pretty sure we went with keyword list, because it's convenient to write, but there was no reason to actually restrict the tags to atoms.


Patch:
diff --git a/lib/kino/control.ex b/lib/kino/control.ex
index 89744ec..f9f676b 100644
--- a/lib/kino/control.ex
+++ b/lib/kino/control.ex
@@ -401,6 +401,8 @@ defmodule Kino.Control do
   @doc """
   Same as `stream/1`, but attaches custom tag to every stream item.
 
+  Tags can be arbitrary terms.
+
   ## Example
 
       button = Kino.Control.button("Hello")
@@ -414,17 +416,17 @@ defmodule Kino.Control do
       #=> {:hello, %{origin: "client1", type: :click}}
       #=> {:check, %{origin: "client1", type: :change, value: true}}
   """
-  @spec tagged_stream(keyword(event_source())) :: Enumerable.t()
+  @spec tagged_stream(list({tag :: term(), event_source()})) :: Enumerable.t()
   def tagged_stream(entries) when is_list(entries) do
     {tagged_topics, tagged_intervals} =
       for entry <- entries, reduce: {[], []} do
         {tagged_topics, tagged_intervals} ->
           case entry do
-            {tag, source} when is_atom(tag) ->
+            {_tag, source} ->
               assert_stream_source!(source)
 
             _other ->
-              raise ArgumentError, "expected a keyword list, got: #{inspect(entries)}"
+              raise ArgumentError, "expected a list of 2-element tuples, got: #{inspect(entries)}"
           end
 
           {tag, source} = entry


Test patch:
diff --git a/test/kino/control_test.exs b/test/kino/control_test.exs
index 56b2ac5..4bc4a26 100644
--- a/test/kino/control_test.exs
+++ b/test/kino/control_test.exs
@@ -165,7 +165,7 @@ defmodule Kino.ControlTest do
 
   describe "tagged_stream/1" do
     test "raises on invalid argument" do
-      assert_raise ArgumentError, "expected a keyword list, got: [0]", fn ->
+      assert_raise ArgumentError, "expected a list of 2-element tuples, got: [0]", fn ->
         Kino.Control.tagged_stream([0])
       end
 
@@ -191,6 +191,16 @@ defmodule Kino.ControlTest do
         |> Enum.take(2)
 
       assert Enum.sort(events) == [{:click, %{origin: "client1"}}, {:name, %{origin: "client2"}}]
+
+      events =
+        [{{:click, "button"}, button}, {{:name, "text"}, input}]
+        |> Kino.Control.tagged_stream()
+        |> Enum.take(2)
+
+      assert Enum.sort(events) == [
+               {{:click, "button"}, %{origin: "client1"}},
+               {{:name, "text"}, %{origin: "client2"}}
+             ]
     end
   end
Repository: livebook-dev/kino
Pull request #492 (2025-02-12 19:40:09)

Fix mermaid output export
It would be nice to have a test for this. Maybe a `Kino.JS.export(Kino.JS.t())` function that would allow us to get the exported value?


Patch:
diff --git a/lib/kino/js/live.ex b/lib/kino/js/live.ex
index d71663e..fce3df4 100644
--- a/lib/kino/js/live.ex
+++ b/lib/kino/js/live.ex
@@ -318,8 +318,8 @@ defmodule Kino.JS.Live do
   end
 
   @doc false
-  @spec output_info(t()) :: map()
-  def output_info(%__MODULE__{} = kino) do
+  @spec output_attrs(t()) :: map()
+  def output_attrs(%__MODULE__{} = kino) do
     %{
       js_view: %{
         ref: kino.ref,
diff --git a/lib/kino/mermaid.ex b/lib/kino/mermaid.ex
index 0390a90..93e2604 100644
--- a/lib/kino/mermaid.ex
+++ b/lib/kino/mermaid.ex
@@ -42,7 +42,7 @@ defmodule Kino.Mermaid do
     Kino.JS.new(
       __MODULE__,
       %{diagram: diagram, caption: opts[:caption], download: opts[:download]},
-      export: fn diagram -> {"mermaid", diagram} end
+      export: fn data -> {"mermaid", data.diagram} end
     )
   end
 end
diff --git a/lib/kino/render.ex b/lib/kino/render.ex
index 817d622..19c3748 100644
--- a/lib/kino/render.ex
+++ b/lib/kino/render.ex
@@ -71,7 +71,7 @@ defimpl Kino.Render, for: Kino.JS.Live do
 
   def to_livebook(kino) do
     Kino.Bridge.reference_object(kino.pid, self())
-    %{js_view: js_view, export: export} = Kino.JS.Live.output_info(kino)
+    %{js_view: js_view, export: export} = Kino.JS.Live.output_attrs(kino)
     %{type: :js, js_view: js_view, export: export}
   end
 end


Test patch:
diff --git a/lib/kino/test.ex b/lib/kino/test.ex
index c4df05c..623553f 100644
--- a/lib/kino/test.ex
+++ b/lib/kino/test.ex
@@ -200,4 +200,29 @@ defmodule Kino.Test do
   def push_smart_cell_editor_source(kino, source) do
     send(kino.pid, {:editor_source, source})
   end
+
+  @doc """
+  Invokes export for the given `Kino.JS` or `Kino.JS.Live` and returns
+  the result.
+
+  For more details, see `Kino.JS.new/3` and `Kino.JS.Live.new/3`
+  respectively.
+  """
+  def export(kino, timeout \\ 100)
+
+  def export(%Kino.JS{} = kino, timeout) do
+    %{js_view: %{ref: ref, pid: pid}} = Kino.JS.output_attrs(kino)
+    do_export(ref, pid, timeout)
+  end
+
+  def export(%Kino.JS.Live{} = kino, timeout) do
+    %{js_view: %{ref: ref, pid: pid}} = Kino.JS.Live.output_attrs(kino)
+    do_export(ref, pid, timeout)
+  end
+
+  defp do_export(ref, pid, timeout) do
+    send(pid, {:export, self(), %{ref: ref}})
+    assert_receive {:export_reply, export_result, %{ref: ^ref}}, timeout
+    export_result
+  end
 end
diff --git a/test/kino/mermaid_test.exs b/test/kino/mermaid_test.exs
new file mode 100644
index 0000000..4f57fcd
--- /dev/null
+++ b/test/kino/mermaid_test.exs
@@ -0,0 +1,17 @@
+defmodule Kino.MermaidTest do
+  use Kino.LivebookCase, async: true
+
+  test "export" do
+    content = """
+    graph TD;
+      A-->B;
+      A-->C;
+      B-->D;
+      C-->D;
+    """
+
+    kino = Kino.Mermaid.new(content)
+
+    assert export(kino) == {"mermaid", content}
+  end
+end