message: relocate from container

This package is quite useful. This change allows it to be imported without importing container.

Signed-off-by: Ophestra <cat@gensokyo.uk>

4 files changed, 539 insertions, 0 deletions

diff --git a/message/message.go b/message/message.go
new file mode 100644
index 00000000..36f8ae4b
--- /dev/null
+++ b/message/message.go
@@ -0,0 +1,111 @@
+// Package message provides interfaces and a base implementation for extended reporting on top of [log.Logger]
+package message
+
+import (
+	"errors"
+	"log"
+	"sync/atomic"
+)
+
+// Error is an error with a user-facing message.
+type Error interface {
+	// Message returns a user-facing error message.
+	Message() string
+
+	error
+}
+
+// GetMessage returns whether an error implements [Error], and the message if it does.
+func GetMessage(err error) (string, bool) {
+	var e Error
+	if !errors.As(err, &e) || e == nil {
+		return "", false
+	}
+	return e.Message(), true
+}
+
+// Msg is used for package-wide verbose logging.
+type Msg interface {
+	// GetLogger returns the address of the underlying [log.Logger].
+	GetLogger() *log.Logger
+
+	// IsVerbose atomically loads and returns whether [Msg] has verbose logging enabled.
+	IsVerbose() bool
+	// SwapVerbose atomically stores a new verbose state and returns the previous value held by [Msg].
+	SwapVerbose(verbose bool) bool
+	// Verbose passes its argument to the Println method of the underlying [log.Logger] if IsVerbose returns true.
+	Verbose(v ...any)
+	// Verbosef passes its argument to the Printf method of the underlying [log.Logger] if IsVerbose returns true.
+	Verbosef(format string, v ...any)
+
+	// Suspend causes the embedded [Suspendable] to withhold writes to its downstream [io.Writer].
+	// Suspend returns false and is a noop if called between calls to Suspend and Resume.
+	Suspend() bool
+
+	// Resume dumps the entire buffer held by the embedded [Suspendable] and stops withholding future writes.
+	// Resume returns false and is a noop if a call to Suspend does not precede it.
+	Resume() bool
+
+	// BeforeExit runs implementation-specific cleanup code, and optionally prints warnings.
+	// BeforeExit is called before [os.Exit].
+	BeforeExit()
+}
+
+// defaultMsg is the default implementation of the [Msg] interface.
+// The zero value is not safe for use. Callers should use the [NewMsg] function instead.
+type defaultMsg struct {
+	verbose atomic.Bool
+
+	logger *log.Logger
+	Suspendable
+}
+
+// NewMsg initialises a downstream [log.Logger] for a new [Msg].
+// The [log.Logger] should no longer be configured after NewMsg returns.
+// If downstream is nil, a new logger is initialised in its place.
+func NewMsg(downstream *log.Logger) Msg {
+	if downstream == nil {
+		downstream = log.New(log.Writer(), "container: ", 0)
+	}
+
+	m := defaultMsg{logger: downstream}
+	m.Suspendable.Downstream = downstream.Writer()
+	downstream.SetOutput(&m.Suspendable)
+	return &m
+}
+
+func (msg *defaultMsg) GetLogger() *log.Logger { return msg.logger }
+
+func (msg *defaultMsg) IsVerbose() bool               { return msg.verbose.Load() }
+func (msg *defaultMsg) SwapVerbose(verbose bool) bool { return msg.verbose.Swap(verbose) }
+func (msg *defaultMsg) Verbose(v ...any) {
+	if msg.verbose.Load() {
+		msg.logger.Println(v...)
+	}
+}
+func (msg *defaultMsg) Verbosef(format string, v ...any) {
+	if msg.verbose.Load() {
+		msg.logger.Printf(format, v...)
+	}
+}
+
+// Resume calls [Suspendable.Resume] and prints a message if buffer was filled
+// between calls to [Suspendable.Suspend] and Resume.
+func (msg *defaultMsg) Resume() bool {
+	resumed, dropped, _, err := msg.Suspendable.Resume()
+	if err != nil {
+		// probably going to result in an error as well, so this message is as good as unreachable
+		msg.logger.Printf("cannot dump buffer on resume: %v", err)
+	}
+	if resumed && dropped > 0 {
+		msg.logger.Printf("dropped %d bytes while output is suspended", dropped)
+	}
+	return resumed
+}
+
+// BeforeExit prints a message if called between calls to [Suspendable.Suspend] and Resume.
+func (msg *defaultMsg) BeforeExit() {
+	if msg.Resume() {
+		msg.logger.Printf("beforeExit reached on suspended output")
+	}
+}
diff --git a/message/message_test.go b/message/message_test.go
new file mode 100644
index 00000000..93ceaacb
--- /dev/null
+++ b/message/message_test.go
@@ -0,0 +1,178 @@
+package message_test
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"log"
+	"strings"
+	"syscall"
+	"testing"
+
+	"hakurei.app/container"
+	"hakurei.app/container/stub"
+	"hakurei.app/message"
+)
+
+func TestMessageError(t *testing.T) {
+	testCases := []struct {
+		name   string
+		err    error
+		want   string
+		wantOk bool
+	}{
+		{"nil", nil, "", false},
+		{"new", errors.New(":3"), "", false},
+		{"start", &container.StartError{
+			Step: "meow",
+			Err:  syscall.ENOTRECOVERABLE,
+		}, "cannot meow: state not recoverable", true},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, ok := message.GetMessage(tc.err)
+			if got != tc.want {
+				t.Errorf("GetMessage: %q, want %q", got, tc.want)
+			}
+			if ok != tc.wantOk {
+				t.Errorf("GetMessage: ok = %v, want %v", ok, tc.wantOk)
+			}
+		})
+	}
+}
+
+func TestDefaultMsg(t *testing.T) {
+	// copied from output.go
+	const suspendBufMax = 1 << 24
+
+	t.Run("logger", func(t *testing.T) {
+		t.Run("nil", func(t *testing.T) {
+			got := message.NewMsg(nil).GetLogger()
+
+			if out := got.Writer().(*message.Suspendable).Downstream; out != log.Writer() {
+				t.Errorf("GetLogger: Downstream = %#v", out)
+			}
+
+			if prefix := got.Prefix(); prefix != "container: " {
+				t.Errorf("GetLogger: prefix = %q", prefix)
+			}
+		})
+
+		t.Run("takeover", func(t *testing.T) {
+			l := log.New(io.Discard, "\x00", 0xdeadbeef)
+			got := message.NewMsg(l)
+
+			if logger := got.GetLogger(); logger != l {
+				t.Errorf("GetLogger: %#v, want %#v", logger, l)
+			}
+
+			if ds := l.Writer().(*message.Suspendable).Downstream; ds != io.Discard {
+				t.Errorf("GetLogger: Downstream = %#v", ds)
+			}
+		})
+	})
+
+	dw := expectWriter{t: t}
+
+	steps := []struct {
+		name     string
+		pt, next []byte
+		err      error
+
+		f func(t *testing.T, msg message.Msg)
+	}{
+		{"zero verbose", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.IsVerbose() {
+				t.Error("IsVerbose unexpected true")
+			}
+		}},
+
+		{"swap false", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.SwapVerbose(false) {
+				t.Error("SwapVerbose unexpected true")
+			}
+		}},
+		{"write discard", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Verbose("\x00")
+			msg.Verbosef("\x00")
+		}},
+		{"verbose false", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.IsVerbose() {
+				t.Error("IsVerbose unexpected true")
+			}
+		}},
+
+		{"swap true", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.SwapVerbose(true) {
+				t.Error("SwapVerbose unexpected true")
+			}
+		}},
+		{"write verbose", []byte("test: \x00\n"), nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Verbose("\x00")
+		}},
+		{"write verbosef", []byte(`test: "\x00"` + "\n"), nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Verbosef("%q", "\x00")
+		}},
+		{"verbose true", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if !msg.IsVerbose() {
+				t.Error("IsVerbose unexpected false")
+			}
+		}},
+
+		{"resume noop", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.Resume() {
+				t.Error("Resume unexpected success")
+			}
+		}},
+		{"beforeExit noop", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.BeforeExit()
+		}},
+
+		{"beforeExit suspend", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Suspend()
+		}},
+		{"beforeExit message", []byte("test: beforeExit reached on suspended output\n"), nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.BeforeExit()
+		}},
+		{"post beforeExit resume noop", nil, nil, nil, func(t *testing.T, msg message.Msg) {
+			if msg.Resume() {
+				t.Error("Resume unexpected success")
+			}
+		}},
+
+		{"suspend", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Suspend()
+		}},
+		{"suspend write", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.GetLogger().Print("\x00")
+		}},
+		{"resume error", []byte("test: \x00\n"), []byte("test: cannot dump buffer on resume: unique error 0 injected by the test suite\n"), stub.UniqueError(0), func(t *testing.T, msg message.Msg) {
+			if !msg.Resume() {
+				t.Error("Resume unexpected failure")
+			}
+		}},
+
+		{"suspend drop", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.Suspend()
+		}},
+		{"suspend write fill", nil, nil, nil, func(_ *testing.T, msg message.Msg) {
+			msg.GetLogger().Print(strings.Repeat("\x00", suspendBufMax))
+		}},
+		{"resume dropped", append([]byte("test: "), bytes.Repeat([]byte{0}, suspendBufMax-6)...), []byte("test: dropped 7 bytes while output is suspended\n"), nil, func(t *testing.T, msg message.Msg) {
+			if !msg.Resume() {
+				t.Error("Resume unexpected failure")
+			}
+		}},
+	}
+
+	msg := message.NewMsg(log.New(&dw, "test: ", 0))
+	for _, step := range steps {
+		// these share the same writer, so cannot be subtests
+		t.Logf("running step %q", step.name)
+		dw.expect, dw.next, dw.err = step.pt, step.next, step.err
+		step.f(t, msg)
+		if dw.expect != nil {
+			t.Errorf("expect: %q", string(dw.expect))
+		}
+	}
+}
diff --git a/message/output.go b/message/output.go
new file mode 100644
index 00000000..cb8df57c
--- /dev/null
+++ b/message/output.go
@@ -0,0 +1,77 @@
+package message
+
+import (
+	"bytes"
+	"io"
+	"sync"
+	"sync/atomic"
+	"syscall"
+)
+
+const (
+	suspendBufInitial = 1 << 12
+	suspendBufMax     = 1 << 24
+)
+
+// Suspendable proxies writes to a downstream [io.Writer] but optionally withholds writes
+// between calls to Suspend and Resume.
+type Suspendable struct {
+	Downstream io.Writer
+
+	s atomic.Bool
+
+	buf bytes.Buffer
+	// for growing buf
+	bufOnce sync.Once
+	// for synchronising all other buf operations
+	bufMu sync.Mutex
+
+	dropped int
+}
+
+func (s *Suspendable) Write(p []byte) (n int, err error) {
+	if !s.s.Load() {
+		return s.Downstream.Write(p)
+	}
+	s.bufOnce.Do(func() { s.buf.Grow(suspendBufInitial) })
+
+	s.bufMu.Lock()
+	defer s.bufMu.Unlock()
+
+	if free := suspendBufMax - s.buf.Len(); free < len(p) {
+		// fast path
+		if free <= 0 {
+			s.dropped += len(p)
+			return 0, syscall.ENOMEM
+		}
+
+		n, _ = s.buf.Write(p[:free])
+		err = syscall.ENOMEM
+		s.dropped += len(p) - n
+		return
+	}
+
+	return s.buf.Write(p)
+}
+
+// IsSuspended returns whether [Suspendable] is currently between a call to Suspend and Resume.
+func (s *Suspendable) IsSuspended() bool { return s.s.Load() }
+
+// Suspend causes [Suspendable] to start withholding output in its buffer.
+func (s *Suspendable) Suspend() bool { return s.s.CompareAndSwap(false, true) }
+
+// Resume undoes the effect of Suspend and dumps the buffered into the downstream [io.Writer].
+func (s *Suspendable) Resume() (resumed bool, dropped uintptr, n int64, err error) {
+	if s.s.CompareAndSwap(true, false) {
+		s.bufMu.Lock()
+		defer s.bufMu.Unlock()
+
+		resumed = true
+		dropped = uintptr(s.dropped)
+
+		s.dropped = 0
+		n, err = io.Copy(s.Downstream, &s.buf)
+		s.buf.Reset()
+	}
+	return
+}
diff --git a/message/output_test.go b/message/output_test.go
new file mode 100644
index 00000000..ab69fe52
--- /dev/null
+++ b/message/output_test.go
@@ -0,0 +1,173 @@
+package message_test
+
+import (
+	"bytes"
+	"errors"
+	"reflect"
+	"strconv"
+	"syscall"
+	"testing"
+
+	"hakurei.app/container/stub"
+	"hakurei.app/message"
+)
+
+func TestSuspendable(t *testing.T) {
+	// copied from output.go
+	const suspendBufMax = 1 << 24
+
+	const (
+		// equivalent to len(want.pt)
+		nSpecialPtEquiv = -iota - 1
+		// equivalent to len(want.w)
+		nSpecialWEquiv
+		// suspends writer before executing test case, implies nSpecialWEquiv
+		nSpecialSuspend
+		// offset: resume writer and measure against dump instead, implies nSpecialPtEquiv
+		nSpecialDump
+	)
+
+	// shares the same writer
+	steps := []struct {
+		name    string
+		w, pt   []byte
+		err     error
+		wantErr error
+		n       int
+	}{
+		{"simple", []byte{0xde, 0xad, 0xbe, 0xef}, []byte{0xde, 0xad, 0xbe, 0xef},
+			nil, nil, nSpecialPtEquiv},
+
+		{"error", []byte{0xb, 0xad}, []byte{0xb, 0xad},
+			stub.UniqueError(0), stub.UniqueError(0), nSpecialPtEquiv},
+
+		{"suspend short", []byte{0}, nil,
+			nil, nil, nSpecialSuspend},
+		{"sw short 0", []byte{0xca, 0xfe, 0xba, 0xbe}, nil,
+			nil, nil, nSpecialWEquiv},
+		{"sw short 1", []byte{0xff}, nil,
+			nil, nil, nSpecialWEquiv},
+		{"resume short", nil, []byte{0, 0xca, 0xfe, 0xba, 0xbe, 0xff}, nil, nil,
+			nSpecialDump},
+
+		{"long pt", bytes.Repeat([]byte{0xff}, suspendBufMax+1), bytes.Repeat([]byte{0xff}, suspendBufMax+1),
+			nil, nil, nSpecialPtEquiv},
+
+		{"suspend fill", bytes.Repeat([]byte{0xfe}, suspendBufMax), nil,
+			nil, nil, nSpecialSuspend},
+		{"drop", []byte{0}, nil,
+			nil, syscall.ENOMEM, 0},
+		{"drop error", []byte{0}, nil,
+			stub.UniqueError(1), syscall.ENOMEM, 0},
+		{"resume fill", nil, bytes.Repeat([]byte{0xfe}, suspendBufMax),
+			nil, nil, nSpecialDump - 2},
+
+		{"suspend fill partial", bytes.Repeat([]byte{0xfd}, suspendBufMax-0xf), nil,
+			nil, nil, nSpecialSuspend},
+		{"partial write", bytes.Repeat([]byte{0xad}, 0x1f), nil,
+			nil, syscall.ENOMEM, 0xf},
+		{"full drop", []byte{0}, nil,
+			nil, syscall.ENOMEM, 0},
+		{"resume fill partial", nil, append(bytes.Repeat([]byte{0xfd}, suspendBufMax-0xf), bytes.Repeat([]byte{0xad}, 0xf)...),
+			nil, nil, nSpecialDump - 0x10 - 1},
+	}
+
+	var dw expectWriter
+
+	w := message.Suspendable{Downstream: &dw}
+	for _, step := range steps {
+		// these share the same writer, so cannot be subtests
+		t.Logf("writing step %q", step.name)
+		dw.expect, dw.err = step.pt, step.err
+
+		var (
+			gotN   int
+			gotErr error
+		)
+
+		wantN := step.n
+		switch wantN {
+		case nSpecialPtEquiv:
+			wantN = len(step.pt)
+			gotN, gotErr = w.Write(step.w)
+
+		case nSpecialWEquiv:
+			wantN = len(step.w)
+			gotN, gotErr = w.Write(step.w)
+
+		case nSpecialSuspend:
+			s := w.IsSuspended()
+			if ok := w.Suspend(); s && ok {
+				t.Fatal("Suspend: unexpected success")
+			}
+
+			wantN = len(step.w)
+			gotN, gotErr = w.Write(step.w)
+
+		default:
+			if wantN <= nSpecialDump {
+				if !w.IsSuspended() {
+					t.Fatal("IsSuspended unexpected false")
+				}
+
+				resumed, dropped, n, err := w.Resume()
+				if !resumed {
+					t.Fatal("Resume: resumed = false")
+				}
+				if wantDropped := nSpecialDump - wantN; int(dropped) != wantDropped {
+					t.Errorf("Resume: dropped = %d, want %d", dropped, wantDropped)
+				}
+
+				wantN = len(step.pt)
+				gotN, gotErr = int(n), err
+			} else {
+				gotN, gotErr = w.Write(step.w)
+			}
+		}
+
+		if gotN != wantN {
+			t.Errorf("Write: n = %d, want %d", gotN, wantN)
+		}
+
+		if !reflect.DeepEqual(gotErr, step.wantErr) {
+			t.Errorf("Write: %v", gotErr)
+		}
+
+		if dw.expect != nil {
+			t.Errorf("expect: %q", string(dw.expect))
+		}
+	}
+}
+
+// expectWriter compares Write calls to expect.
+type expectWriter struct {
+	expect []byte
+	err    error
+
+	// optional consecutive write
+	next []byte
+
+	// optional, calls Error on failure if not nil
+	t *testing.T
+}
+
+func (w *expectWriter) Write(p []byte) (n int, err error) {
+	defer func() { w.expect = w.next; w.next = nil }()
+
+	n, err = len(p), w.err
+	if w.expect == nil {
+		n, err = 0, errors.New("unexpected call to Write: "+strconv.Quote(string(p)))
+		if w.t != nil {
+			w.t.Error(err.Error())
+		}
+		return
+	}
+	if string(p) != string(w.expect) {
+		n, err = 0, errors.New("p = "+strconv.Quote(string(p))+", want "+strconv.Quote(string(w.expect)))
+		if w.t != nil {
+			w.t.Error(err.Error())
+		}
+		return
+	}
+	return
+}