From a788a56e2e0b5896f8315f439c97247722b1fcac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Thomas=20Kj=C3=A6r=20Aabo?= Date: Fri, 30 Oct 2026 10:45:29 +0100 Subject: [PATCH 1/2] BUG/MEDIUM: runtime: preserve map payload entries across chunks parseMapPayload could lose an entry when it crossed a chunk boundary, lose deferred data after repeated crossings, or repeat stale data when a chunk exactly matched the buffer size. Build chunks with one buffer and flush it before adding an entry that would exceed the limit. This preserves every entry in its original order while keeping the existing AddMapPayloadVersioned size behavior. --- runtime/runtime_client.go | 30 +++++++++------------------ runtime/runtime_client_test.go | 37 ++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 20 deletions(-) diff --git a/runtime/runtime_client.go b/runtime/runtime_client.go index 92e8d799..b4fcb1b0 100644 --- a/runtime/runtime_client.go +++ b/runtime/runtime_client.go @@ -694,37 +694,27 @@ func (c *client) AddMapPayload(name, payload string) error { } func parseMapPayload(entries models.MapEntries, maxBufSize int) (bool, []string) { - prevKV := "" - currKV := "" data := "" var payload []string var exceededSize bool for _, d := range entries { - if prevKV != "" { - data += prevKV - prevKV = "" - } kv := d.Key + " " + d.Value + "\n" - data += kv - switch { - case len(data) < maxBufSize: - currKV = data - case len(data) == maxBufSize: + if data != "" && len(data)+len(kv) > maxBufSize { + exceededSize = true payload = append(payload, data) data = "" - case len(data) > maxBufSize: - exceededSize = true - if currKV == "" { - currKV = kv + } + data += kv + if len(data) >= maxBufSize { + if len(data) > maxBufSize { + exceededSize = true } - payload = append(payload, currKV) - prevKV = d.Key + " " + d.Value + "\n" + payload = append(payload, data) data = "" - currKV = "" } } - if len(currKV) > 0 { - payload = append(payload, currKV) + if data != "" { + payload = append(payload, data) } return exceededSize, payload } diff --git a/runtime/runtime_client_test.go b/runtime/runtime_client_test.go index 8e672246..c5f363c0 100644 --- a/runtime/runtime_client_test.go +++ b/runtime/runtime_client_test.go @@ -65,6 +65,43 @@ func Test_parseMapPayload(t *testing.T) { "key2 val2\n", }, }, + { + name: "Keep final entry after crossing maxBufSize", + args: args{entries: models.MapEntries{ + &models.MapEntry{Key: "a", Value: "bbbb"}, + &models.MapEntry{Key: "b", Value: "bbbb"}, + }, maxBufSize: 10}, + wantExceededSize: true, + wantPayload: []string{ + "a bbbb\n", + "b bbbb\n", + }, + }, + { + name: "Keep entries across repeated maxBufSize crossings", + args: args{entries: models.MapEntries{ + &models.MapEntry{Key: "a", Value: "bbbb"}, + &models.MapEntry{Key: "b", Value: "bbbb"}, + &models.MapEntry{Key: "c", Value: "bbbb"}, + }, maxBufSize: 10}, + wantExceededSize: true, + wantPayload: []string{ + "a bbbb\n", + "b bbbb\n", + "c bbbb\n", + }, + }, + { + name: "Do not repeat data after reaching maxBufSize exactly", + args: args{entries: models.MapEntries{ + &models.MapEntry{Key: "a", Value: "b"}, + &models.MapEntry{Key: "c", Value: "ddd"}, + }, maxBufSize: 10}, + wantExceededSize: false, + wantPayload: []string{ + "a b\nc ddd\n", + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { From b0d7d130095fcecf0eb193481280976510a8799a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Thomas=20Kj=C3=A6r=20Aabo?= Date: Fri, 30 Oct 2026 10:46:51 +0100 Subject: [PATCH 2/2] MINOR: runtime: add atomic map replacement Add ReplaceMapPayload to prepare a new map version, clear it, add the desired entries in order, and commit only after every step succeeds. Require HAProxy 2.4 or newer instead of falling back to a non-atomic update. An empty entry list commits an empty map. --- runtime/interface.go | 2 + runtime/runtime_client.go | 32 ++++++ runtime/runtime_client_test.go | 203 +++++++++++++++++++++++++++++++++ 3 files changed, 237 insertions(+) diff --git a/runtime/interface.go b/runtime/interface.go index 1d13b276..af17736a 100644 --- a/runtime/interface.go +++ b/runtime/interface.go @@ -52,6 +52,8 @@ type Maps interface { // ParseMapEntriesFromFile reads entries from file ParseMapEntriesFromFile(inputFile io.Reader, hasID bool) models.MapEntries AddMapPayloadVersioned(name string, entries models.MapEntries) error + // ReplaceMapPayload atomically replaces all entries in a map. + ReplaceMapPayload(name string, entries models.MapEntries) error // PrepareMap allocates a new map version PrepareMap(name string) (version string, err error) // CommitMap commits all changes made to a map version diff --git a/runtime/runtime_client.go b/runtime/runtime_client.go index b4fcb1b0..0d5a4fa3 100644 --- a/runtime/runtime_client.go +++ b/runtime/runtime_client.go @@ -760,6 +760,38 @@ func (c *client) AddMapPayloadVersioned(name string, entries models.MapEntries) return nil } +// ReplaceMapPayload atomically replaces all entries in a map using a prepared map version. +func (c *client) ReplaceMapPayload(name string, entries models.MapEntries) error { + if !c.runtime.IsValid() { + return errors.New("no valid runtime found") + } + v := HAProxyVersion{Version: semver.MustParse("2.4")} + if !c.IsVersionBiggerOrEqual(&v) { + return fmt.Errorf("not supported for HAProxy versions lower than 2.4 %w", native_errors.ErrGeneral) + } + name, err := c.GetMapsPath(name) + if err != nil { + return fmt.Errorf("%s %w", c.runtime.socketPath, err) + } + _, payload := parseMapPayload(entries, maxBufSize) + version, err := c.runtime.PrepareMap(name) + if err != nil { + return fmt.Errorf("%s %w", c.runtime.socketPath, err) + } + if err = c.runtime.ClearMapVersioned(name, version); err != nil { + return fmt.Errorf("%s %w", c.runtime.socketPath, err) + } + for i := range payload { + if err = c.runtime.AddMapPayloadVersioned(version, name, payload[i]); err != nil { + return fmt.Errorf("%s %w", c.runtime.socketPath, err) + } + } + if err = c.runtime.CommitMap(version, name); err != nil { + return fmt.Errorf("%s %w", c.runtime.socketPath, err) + } + return nil +} + // AddMapEntry adds an entry into the map file func (c *client) AddMapEntry(name, key, value string) error { if !c.runtime.IsValid() { diff --git a/runtime/runtime_client_test.go b/runtime/runtime_client_test.go index c5f363c0..2e5cc222 100644 --- a/runtime/runtime_client_test.go +++ b/runtime/runtime_client_test.go @@ -16,12 +16,119 @@ package runtime import ( + "bufio" + "errors" + "net" + "os" + "strings" + "sync" "testing" + "github.com/Masterminds/semver/v3" + native_errors "github.com/haproxytech/client-native/v6/errors" "github.com/haproxytech/client-native/v6/models" + "github.com/haproxytech/client-native/v6/runtime/options" "github.com/stretchr/testify/assert" ) +type mapCommandSocket struct { + net.Listener + + reply func(string) string + mu sync.Mutex + commands []string +} + +func newMapCommandSocket(t *testing.T, reply func(string) string) *mapCommandSocket { + t.Helper() + f, err := os.CreateTemp("", "client-native-map-*") + if err != nil { + t.Fatal(err) + } + addr := f.Name() + _ = f.Close() + _ = os.Remove(addr) + listener, err := net.Listen("unix", addr) + if err != nil { + t.Fatal(err) + } + socket := &mapCommandSocket{Listener: listener, reply: reply} + go socket.serve() + t.Cleanup(func() { + _ = listener.Close() + _ = os.Remove(addr) + }) + return socket +} + +func (s *mapCommandSocket) serve() { + for { + conn, err := s.Accept() + if err != nil { + return + } + go s.handleConnection(conn) + } +} + +func (s *mapCommandSocket) handleConnection(conn net.Conn) { + defer conn.Close() + + reader := bufio.NewReader(conn) + command, err := reader.ReadString('\n') + if err != nil { + return + } + if strings.Contains(command, "<<") { + for { + line, readErr := reader.ReadString('\n') + command += line + if line == "\n" || readErr != nil { + break + } + } + } + command = strings.TrimPrefix(command, "set severity-output number;") + command = strings.TrimSuffix(command, "\n") + + s.mu.Lock() + s.commands = append(s.commands, command) + s.mu.Unlock() + + response := "" + if s.reply != nil { + response = s.reply(command) + } + if response == "" && strings.HasPrefix(command, "prepare map ") { + response = "map transaction: version: 1" + } + _, _ = conn.Write([]byte(response)) +} + +func (s *mapCommandSocket) Commands() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.commands...) +} + +func newMapClient(t *testing.T, socket *mapCommandSocket, version string) *client { + t.Helper() + runtime := &SingleRuntime{} + if err := runtime.Init(socket.Addr().String(), false, options.RuntimeOptions{DoNotCheckRuntimeOnInit: true}); err != nil { + t.Fatal(err) + } + mapsDir := "/etc/haproxy/maps" + previousVersion := haproxyVersion + haproxyVersion = &HAProxyVersion{Version: semver.MustParse(version)} + t.Cleanup(func() { + haproxyVersion = previousVersion + }) + return &client{ + options: options.RuntimeOptions{MapsDir: &mapsDir}, + runtime: runtime, + } +} + func Test_parseMapPayload(t *testing.T) { type args struct { entries models.MapEntries @@ -115,3 +222,99 @@ func Test_parseMapPayload(t *testing.T) { }) } } + +func TestReplaceMapPayload(t *testing.T) { + socket := newMapCommandSocket(t, nil) + client := newMapClient(t, socket, "2.4.0") + entries := models.MapEntries{ + &models.MapEntry{Key: "first", Value: "one"}, + &models.MapEntry{Key: "second", Value: "two"}, + } + + if err := client.ReplaceMapPayload("example", entries); err != nil { + t.Fatalf("ReplaceMapPayload() error = %v", err) + } + + path := "/etc/haproxy/maps/example.map" + assert.Equal(t, []string{ + "prepare map " + path, + "clear map @1 " + path, + "add map @1 " + path + " <<\nfirst one\nsecond two\n", + "commit map @1 " + path, + }, socket.Commands()) +} + +func TestReplaceMapPayloadMultipleChunksPreservesOrder(t *testing.T) { + socket := newMapCommandSocket(t, nil) + client := newMapClient(t, socket, "2.4.0") + firstValue := strings.Repeat("a", maxBufSize-len("first")-2) + secondValue := strings.Repeat("b", maxBufSize-len("second")-2) + entries := models.MapEntries{ + &models.MapEntry{Key: "first", Value: firstValue}, + &models.MapEntry{Key: "second", Value: secondValue}, + } + + if err := client.ReplaceMapPayload("example", entries); err != nil { + t.Fatalf("ReplaceMapPayload() error = %v", err) + } + + path := "/etc/haproxy/maps/example.map" + assert.Equal(t, []string{ + "prepare map " + path, + "clear map @1 " + path, + "add map @1 " + path + " <<\nfirst " + firstValue + "\n", + "add map @1 " + path + " <<\nsecond " + secondValue + "\n", + "commit map @1 " + path, + }, socket.Commands()) +} + +func TestReplaceMapPayloadEmpty(t *testing.T) { + socket := newMapCommandSocket(t, nil) + client := newMapClient(t, socket, "2.4.0") + + if err := client.ReplaceMapPayload("example", nil); err != nil { + t.Fatalf("ReplaceMapPayload() error = %v", err) + } + + path := "/etc/haproxy/maps/example.map" + assert.Equal(t, []string{ + "prepare map " + path, + "clear map @1 " + path, + "commit map @1 " + path, + }, socket.Commands()) +} + +func TestReplaceMapPayloadUnsupported(t *testing.T) { + socket := newMapCommandSocket(t, nil) + client := newMapClient(t, socket, "2.3.0") + + err := client.ReplaceMapPayload("example", nil) + if !errors.Is(err, native_errors.ErrGeneral) { + t.Fatalf("ReplaceMapPayload() error = %v, want %v", err, native_errors.ErrGeneral) + } + assert.Empty(t, socket.Commands()) +} + +func TestReplaceMapPayloadDoesNotCommitAfterAddFailure(t *testing.T) { + socket := newMapCommandSocket(t, func(command string) string { + if strings.HasPrefix(command, "add map ") { + return "[3]: add failed" + } + return "" + }) + client := newMapClient(t, socket, "2.4.0") + + err := client.ReplaceMapPayload("example", models.MapEntries{ + &models.MapEntry{Key: "first", Value: "one"}, + }) + if !errors.Is(err, native_errors.ErrGeneral) { + t.Fatalf("ReplaceMapPayload() error = %v, want %v", err, native_errors.ErrGeneral) + } + + path := "/etc/haproxy/maps/example.map" + assert.Equal(t, []string{ + "prepare map " + path, + "clear map @1 " + path, + "add map @1 " + path + " <<\nfirst one\n", + }, socket.Commands()) +}