feat(rayapp) Anyscale CLI pr0 add tailWriter with lazy double-buffer implementation

rayapp/tail_writer.go

@@ -0,0 +1,72 @@
+package rayapp
+
+import "bytes"
+
+// tailWriter keeps the most recent `limit` bytes written to it.
+// It uses a double-buffer strategy with two bytes.Buffers: writes go
+// into `active`; when active exceeds half the limit, the older buffer
+// is discarded and the two are swapped. Initial memory footprint is
+// near zero because bytes.Buffer grows lazily.
+type tailWriter struct {
+	stale  bytes.Buffer
+	active bytes.Buffer
+	limit  int
+	half   int
+}
+
+func newTailWriter(limit int) *tailWriter {
+	return &tailWriter{
+		limit: limit,
+		half:  limit / 2,
+	}
+}
+
+func (w *tailWriter) Write(p []byte) (int, error) {
+	n := len(p)
+	if n == 0 {
+		return 0, nil
+	}
+
+	// If a single write is >= limit, keep only the last `limit` bytes.
+	if n >= w.limit {
+		w.stale.Reset()
+		w.active.Reset()
+		w.active.Write(p[n-w.limit:])
+		return n, nil
+	}
+
+	w.active.Write(p)
+
+	if w.active.Len() > w.half {
+		w.rotate()
+	}
+	return n, nil
+}
+
+func (w *tailWriter) rotate() {
+	w.stale.Reset()
+	w.stale, w.active = w.active, w.stale
+}
+
+// String returns the most recent `limit` bytes as a string.
+func (w *tailWriter) String() string {
+	staleBytes := w.stale.Bytes()
+	activeBytes := w.active.Bytes()
+	total := len(staleBytes) + len(activeBytes)
+
+	if total <= w.limit {
+		var buf bytes.Buffer
+		buf.Grow(total)
+		buf.Write(staleBytes)
+		buf.Write(activeBytes)
+		return buf.String()
+	}
+
+	// Combined exceeds limit — trim from the front of stale.
+	skip := total - w.limit
+	var buf bytes.Buffer
+	buf.Grow(w.limit)
+	buf.Write(staleBytes[skip:])
+	buf.Write(activeBytes)
+	return buf.String()
+}

rayapp/tail_writer_test.go

@@ -0,0 +1,84 @@
+package rayapp
+
+import "testing"
+
+func TestTailWriter(t *testing.T) {
+	t.Run("under limit", func(t *testing.T) {
+		tw := newTailWriter(16)
+		tw.Write([]byte("hello"))
+		if got, want := tw.String(), "hello"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("multiple writes under limit", func(t *testing.T) {
+		tw := newTailWriter(16)
+		tw.Write([]byte("abc"))
+		tw.Write([]byte("def"))
+		if got, want := tw.String(), "abcdef"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("exact limit", func(t *testing.T) {
+		tw := newTailWriter(5)
+		tw.Write([]byte("abcde"))
+		if got, want := tw.String(), "abcde"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("wraps around keeps tail", func(t *testing.T) {
+		tw := newTailWriter(5)
+		tw.Write([]byte("abc"))
+		tw.Write([]byte("defgh"))
+		if got, want := tw.String(), "defgh"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("multiple wraps", func(t *testing.T) {
+		tw := newTailWriter(4)
+		tw.Write([]byte("ab"))
+		tw.Write([]byte("cd"))
+		tw.Write([]byte("ef"))
+		if got, want := tw.String(), "cdef"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("single write exceeds limit", func(t *testing.T) {
+		tw := newTailWriter(4)
+		n, err := tw.Write([]byte("abcdefgh"))
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if n != 8 {
+			t.Errorf("Write() = %d, want 8", n)
+		}
+		if got, want := tw.String(), "efgh"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("write returns full length", func(t *testing.T) {
+		tw := newTailWriter(4)
+		tw.Write([]byte("abc"))
+		n, err := tw.Write([]byte("defgh"))
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if n != 5 {
+			t.Errorf("Write() = %d, want 5", n)
+		}
+	})
+
+	t.Run("empty write", func(t *testing.T) {
+		tw := newTailWriter(4)
+		tw.Write([]byte("ab"))
+		tw.Write([]byte(""))
+		if got, want := tw.String(), "ab"; got != want {
+			t.Errorf("String() = %q, want %q", got, want)
+		}
+	})
+}