feat(statefile): write the state file at most once per second

commit: b996bb592d603cdd3e0d0cacf8fd2543852a5f74
author: brice · 40h ago · 2026-10-21 07:10 UTC
committer: brice · 40h ago · 2026-10-21 07:13 UTC
parent: f8e0ac9
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +105 -5

M internal/adapters/statefile/statefile.go +40 -5


@@ -12,6 +12,7 @@ import (
12	12		"path/filepath"
13	13		"strconv"
14	14		"strings"
	15	+	"time"
15	16
16	17		"github.com/bnema/neferwl/internal/adapters/workspaceid"
17	18		"github.com/bnema/neferwl/internal/ports"

@@ -141,8 +142,13 @@ func window(w ports.WindowState, ids *workspaceid.IDs) Window {
141	142		return Window{ID: uint64(w.ID), AppID: w.AppID, PID: w.PID, Output: w.Output, Workspace: w.Workspace, WorkspaceID: ids.ID(w.WorkspaceID, w.WorkspaceName), Visible: w.Visible, Floating: w.Floating, StashIndex: w.StashIndex, StashCount: w.StashCount, Column: w.Column, Row: w.Row, Hidden: w.Hidden}
142	143		}
143	144
144		-	// Run writes every state it receives to path until ctx ends, then removes
145		-	// the file: a stale state must not outlive the session.
	145	+	// writeInterval is the least time between two writes of the file.
	146	+	const writeInterval = time.Second
	147	+
	148	+	// Run writes the states it receives to path until ctx ends, then removes
	149	+	// the file: a stale state must not outlive the session. It writes at most
	150	+	// once per writeInterval: a state after a quiet interval is written at once,
	151	+	// others are coalesced and the latest is written when the interval ends.
146	152		func Run(ctx context.Context, path string, states <-chan ports.State, ids *workspaceid.IDs, log zerowrap.Logger) error {
147	153		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
148	154		return err

@@ -150,15 +156,44 @@ func Run(ctx context.Context, path string, states <-chan ports.State, ids *works
150	156		// A file left by a crashed session on this socket is not ours.
151	157		_ = os.Remove(path)
152	158		defer os.Remove(path)
	159	+	flush := func(st ports.State) {
	160	+	if err := write(path, fromPorts(st, ids)); err != nil {
	161	+	// Scripts lose their state; the session goes on.
	162	+	log.Warn().Err(err).Str("path", path).Msg("write state")
	163	+	}
	164	+	}
	165	+	timer := time.NewTimer(0)
	166	+	defer timer.Stop()
	167	+	<-timer.C
	168	+	var (
	169	+	last time.Time
	170	+	pending ports.State
	171	+	waiting bool
	172	+	)
153	173		for {
	174	+	var due <-chan time.Time
	175	+	if waiting {
	176	+	due = timer.C
	177	+	}
154	178		select {
155	179		case <-ctx.Done():
156	180		return nil
157	181		case st := <-states:
158		-	if err := write(path, fromPorts(st, ids)); err != nil {
159		-	// Scripts lose their state; the session goes on.
160		-	log.Warn().Err(err).Str("path", path).Msg("write state")
	182	+	if waiting {
	183	+	pending = st
	184	+	continue
	185	+	}
	186	+	if wait := writeInterval - time.Since(last); !last.IsZero() && wait > 0 {
	187	+	pending, waiting = st, true
	188	+	timer.Reset(wait)
	189	+	continue
161	190		}
	191	+	flush(st)
	192	+	last = time.Now()
	193	+	case <-due:
	194	+	waiting = false
	195	+	flush(pending)
	196	+	last = time.Now()
162	197		}
163	198		}
164	199		}


@@ -12,6 +12,7 @@ import (
12		"path/filepath"	12		"path/filepath"
13		"strconv"	13		"strconv"
14		"strings"	14		"strings"
			15	+	"time"
15			16
16		"github.com/bnema/neferwl/internal/adapters/workspaceid"	17		"github.com/bnema/neferwl/internal/adapters/workspaceid"
17		"github.com/bnema/neferwl/internal/ports"	18		"github.com/bnema/neferwl/internal/ports"

@@ -141,8 +142,13 @@ func window(w ports.WindowState, ids *workspaceid.IDs) Window {
141		return Window{ID: uint64(w.ID), AppID: w.AppID, PID: w.PID, Output: w.Output, Workspace: w.Workspace, WorkspaceID: ids.ID(w.WorkspaceID, w.WorkspaceName), Visible: w.Visible, Floating: w.Floating, StashIndex: w.StashIndex, StashCount: w.StashCount, Column: w.Column, Row: w.Row, Hidden: w.Hidden}	142		return Window{ID: uint64(w.ID), AppID: w.AppID, PID: w.PID, Output: w.Output, Workspace: w.Workspace, WorkspaceID: ids.ID(w.WorkspaceID, w.WorkspaceName), Visible: w.Visible, Floating: w.Floating, StashIndex: w.StashIndex, StashCount: w.StashCount, Column: w.Column, Row: w.Row, Hidden: w.Hidden}
142		}	143		}
143			144
144	-	// Run writes every state it receives to path until ctx ends, then removes	145	+	// writeInterval is the least time between two writes of the file.
145	-	// the file: a stale state must not outlive the session.	146	+	const writeInterval = time.Second
			147	+
			148	+	// Run writes the states it receives to path until ctx ends, then removes
			149	+	// the file: a stale state must not outlive the session. It writes at most
			150	+	// once per writeInterval: a state after a quiet interval is written at once,
			151	+	// others are coalesced and the latest is written when the interval ends.
146		func Run(ctx context.Context, path string, states <-chan ports.State, ids *workspaceid.IDs, log zerowrap.Logger) error {	152		func Run(ctx context.Context, path string, states <-chan ports.State, ids *workspaceid.IDs, log zerowrap.Logger) error {
147		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	153		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
148		return err	154		return err

@@ -150,15 +156,44 @@ func Run(ctx context.Context, path string, states <-chan ports.State, ids *works
150		// A file left by a crashed session on this socket is not ours.	156		// A file left by a crashed session on this socket is not ours.
151		_ = os.Remove(path)	157		_ = os.Remove(path)
152		defer os.Remove(path)	158		defer os.Remove(path)
			159	+	flush := func(st ports.State) {
			160	+	if err := write(path, fromPorts(st, ids)); err != nil {
			161	+	// Scripts lose their state; the session goes on.
			162	+	log.Warn().Err(err).Str("path", path).Msg("write state")
			163	+	}
			164	+	}
			165	+	timer := time.NewTimer(0)
			166	+	defer timer.Stop()
			167	+	<-timer.C
			168	+	var (
			169	+	last time.Time
			170	+	pending ports.State
			171	+	waiting bool
			172	+	)
153		for {	173		for {
			174	+	var due <-chan time.Time
			175	+	if waiting {
			176	+	due = timer.C
			177	+	}
154		select {	178		select {
155		case <-ctx.Done():	179		case <-ctx.Done():
156		return nil	180		return nil
157		case st := <-states:	181		case st := <-states:
158	-	if err := write(path, fromPorts(st, ids)); err != nil {	182	+	if waiting {
159	-	// Scripts lose their state; the session goes on.	183	+	pending = st
160	-	log.Warn().Err(err).Str("path", path).Msg("write state")	184	+	continue
			185	+	}
			186	+	if wait := writeInterval - time.Since(last); !last.IsZero() && wait > 0 {
			187	+	pending, waiting = st, true
			188	+	timer.Reset(wait)
			189	+	continue
161		}	190		}
			191	+	flush(st)
			192	+	last = time.Now()
			193	+	case <-due:
			194	+	waiting = false
			195	+	flush(pending)
			196	+	last = time.Now()
162		}	197		}
163		}	198		}
164		}	199		}

M internal/adapters/statefile/statefile_test.go +65 -0


@@ -9,6 +9,7 @@ import (
9	9		"reflect"
10	10		"strings"
11	11		"testing"
	12	+	"testing/synctest"
12	13		"time"
13	14
14	15		"github.com/bnema/neferwl/internal/adapters/workspaceid"

@@ -187,3 +188,67 @@ func TestWorkspaceGeometryFields(t *testing.T) {
187	188		}
188	189		}
189	190		}
	191	+
	192	+	// The file is written at most once per second: the first state and one after
	193	+	// a quiet second at once, a burst within the second as one trailing write of
	194	+	// the latest state.
	195	+	func TestRunRateLimit(t *testing.T) {
	196	+	synctest.Test(t, func(t *testing.T) {
	197	+	path := filepath.Join(t.TempDir(), "s", "wayland-9.json")
	198	+	states := make(chan ports.State)
	199	+	ctx, cancel := context.WithCancel(context.Background())
	200	+	done := make(chan error, 1)
	201	+	go func() { done <- Run(ctx, path, states, workspaceid.WithPrefix("0a1b2c3d"), zerowrap.Default()) }()
	202	+	output := func() string {
	203	+	t.Helper()
	204	+	synctest.Wait()
	205	+	st, err := Read(path)
	206	+	if err != nil {
	207	+	return ""
	208	+	}
	209	+	return st.Output
	210	+	}
	211	+	send := func(name string) { states <- ports.State{Output: name} }
	212	+
	213	+	send("a")
	214	+	if got := output(); got != "a" {
	215	+	t.Fatalf("first state not written at once: %q", got)
	216	+	}
	217	+	// A burst within the second: nothing yet, then the latest only.
	218	+	time.Sleep(200 * time.Millisecond)
	219	+	send("b")
	220	+	send("c")
	221	+	send("d")
	222	+	if got := output(); got != "a" {
	223	+	t.Fatalf("written within the second: %q", got)
	224	+	}
	225	+	time.Sleep(799 * time.Millisecond)
	226	+	if got := output(); got != "a" {
	227	+	t.Fatalf("written before the second ended: %q", got)
	228	+	}
	229	+	time.Sleep(time.Millisecond)
	230	+	if got := output(); got != "d" {
	231	+	t.Fatalf("trailing write: %q, want d", got)
	232	+	}
	233	+	// Nothing more is pending: the file stays as it is.
	234	+	if err := os.Remove(path); err != nil {
	235	+	t.Fatal(err)
	236	+	}
	237	+	time.Sleep(5 * time.Second)
	238	+	if got := output(); got != "" {
	239	+	t.Fatalf("a write with no new state: %q", got)
	240	+	}
	241	+	// After a quiet period the next state is immediate.
	242	+	send("e")
	243	+	if got := output(); got != "e" {
	244	+	t.Fatalf("state after a quiet period: %q", got)
	245	+	}
	246	+	cancel()
	247	+	if err := <-done; err != nil {
	248	+	t.Fatal(err)
	249	+	}
	250	+	if _, err := os.Stat(path); !os.IsNotExist(err) {
	251	+	t.Fatal("state file left behind", err)
	252	+	}
	253	+	})
	254	+	}


@@ -9,6 +9,7 @@ import (
9	"reflect"	9		"reflect"
10	"strings"	10		"strings"
11	"testing"	11		"testing"
		12	+	"testing/synctest"
12	"time"	13		"time"
13		14
14	"github.com/bnema/neferwl/internal/adapters/workspaceid"	15		"github.com/bnema/neferwl/internal/adapters/workspaceid"

@@ -187,3 +188,67 @@ func TestWorkspaceGeometryFields(t *testing.T) {
187	}	188		}
188	}	189		}
189	}	190		}
		191	+
		192	+	// The file is written at most once per second: the first state and one after
		193	+	// a quiet second at once, a burst within the second as one trailing write of
		194	+	// the latest state.
		195	+	func TestRunRateLimit(t *testing.T) {
		196	+	synctest.Test(t, func(t *testing.T) {
		197	+	path := filepath.Join(t.TempDir(), "s", "wayland-9.json")
		198	+	states := make(chan ports.State)
		199	+	ctx, cancel := context.WithCancel(context.Background())
		200	+	done := make(chan error, 1)
		201	+	go func() { done <- Run(ctx, path, states, workspaceid.WithPrefix("0a1b2c3d"), zerowrap.Default()) }()
		202	+	output := func() string {
		203	+	t.Helper()
		204	+	synctest.Wait()
		205	+	st, err := Read(path)
		206	+	if err != nil {
		207	+	return ""
		208	+	}
		209	+	return st.Output
		210	+	}
		211	+	send := func(name string) { states <- ports.State{Output: name} }
		212	+
		213	+	send("a")
		214	+	if got := output(); got != "a" {
		215	+	t.Fatalf("first state not written at once: %q", got)
		216	+	}
		217	+	// A burst within the second: nothing yet, then the latest only.
		218	+	time.Sleep(200 * time.Millisecond)
		219	+	send("b")
		220	+	send("c")
		221	+	send("d")
		222	+	if got := output(); got != "a" {
		223	+	t.Fatalf("written within the second: %q", got)
		224	+	}
		225	+	time.Sleep(799 * time.Millisecond)
		226	+	if got := output(); got != "a" {
		227	+	t.Fatalf("written before the second ended: %q", got)
		228	+	}
		229	+	time.Sleep(time.Millisecond)
		230	+	if got := output(); got != "d" {
		231	+	t.Fatalf("trailing write: %q, want d", got)
		232	+	}
		233	+	// Nothing more is pending: the file stays as it is.
		234	+	if err := os.Remove(path); err != nil {
		235	+	t.Fatal(err)
		236	+	}
		237	+	time.Sleep(5 * time.Second)
		238	+	if got := output(); got != "" {
		239	+	t.Fatalf("a write with no new state: %q", got)
		240	+	}
		241	+	// After a quiet period the next state is immediate.
		242	+	send("e")
		243	+	if got := output(); got != "e" {
		244	+	t.Fatalf("state after a quiet period: %q", got)
		245	+	}
		246	+	cancel()
		247	+	if err := <-done; err != nil {
		248	+	t.Fatal(err)
		249	+	}
		250	+	if _, err := os.Stat(path); !os.IsNotExist(err) {
		251	+	t.Fatal("state file left behind", err)
		252	+	}
		253	+	})
		254	+	}