Expose skipped ownership conflicts during PostgreSQL pushes

Author: rodboev

cmd/agentsview/pg.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"os/signal"
@@ -112,27 +113,60 @@ func runPGPush(cfg PGPushConfig) {
 	fmt.Println("Starting PostgreSQL push...")
 	result, err := ps.Push(ctx, forceFull,
 		func(p postgres.PushProgress) {
-			fmt.Printf(
-				"\rPushing... %d/%d sessions, %d messages",
-				p.SessionsDone, p.SessionsTotal,
-				p.MessagesDone,
-			)
+			printPGPushProgress(p)
 		},
 	)
 	fmt.Print("\r\033[K") // clear progress line
 	if err != nil {
 		fatal("pg push: %v", err)
 	}
+	writePGPushSummary(os.Stdout, result)
+	if result.Errors > 0 {
+		fatal("pg push: %d session(s) failed",
+			result.Errors)
+	}
+}
+
+func printPGPushProgress(p postgres.PushProgress) {
+	if p.SkippedConflicts > 0 {
+		fmt.Printf(
+			"\rPushing... %d/%d sessions, %d messages, %d ownership conflicts skipped",
+			p.SessionsDone, p.SessionsTotal,
+			p.MessagesDone, p.SkippedConflicts,
+		)
+		return
+	}
 	fmt.Printf(
+		"\rPushing... %d/%d sessions, %d messages",
+		p.SessionsDone, p.SessionsTotal,
+		p.MessagesDone,
+	)
+}
+
+func writePGPushSummary(w io.Writer, result postgres.PushResult) {
+	if result.SkippedConflicts > 0 {
+		fmt.Fprintf(
+			w,
+			"Pushed %d sessions, %d messages, skipped %d ownership conflict(s) in %s\n",
+			result.SessionsPushed,
+			result.MessagesPushed,
+			result.SkippedConflicts,
+			result.Duration.Round(time.Millisecond),
+		)
+		fmt.Fprintf(
+			w,
+			"Warning: skipped %d session(s) owned by another PostgreSQL push marker\n",
+			result.SkippedConflicts,
+		)
+		return
+	}
+	fmt.Fprintf(
+		w,
 		"Pushed %d sessions, %d messages in %s\n",
 		result.SessionsPushed,
 		result.MessagesPushed,
 		result.Duration.Round(time.Millisecond),
 	)
-	if result.Errors > 0 {
-		fatal("pg push: %d session(s) failed",
-			result.Errors)
-	}
 }
 
 func runPGStatus() {

cmd/agentsview/pg_test.go

@@ -1,15 +1,18 @@
 package main
 
 import (
+	"bytes"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/postgres"
 )
 
 func loadPGServeConfigForTest(t *testing.T, args ...string) (config.Config, string, error) {
@@ -166,3 +169,20 @@ func TestRunPGServeHelperProcess(t *testing.T) {
 	require.NoError(t, err)
 	runPGServe(cfg, basePath)
 }
+
+func TestWritePGPushSummaryIncludesSkippedConflicts(t *testing.T) {
+	var out bytes.Buffer
+
+	writePGPushSummary(&out, postgres.PushResult{
+		SessionsPushed:   3,
+		MessagesPushed:   9,
+		SkippedConflicts: 2,
+		Duration:         1500 * time.Millisecond,
+	})
+
+	got := out.String()
+	assert.Contains(t, got,
+		"Pushed 3 sessions, 9 messages, skipped 2 ownership conflict(s) in 1.5s")
+	assert.Contains(t, got,
+		"Warning: skipped 2 session(s) owned by another PostgreSQL push marker")
+}

cmd/agentsview/pg_watch.go

@@ -64,21 +64,42 @@ func (p *pgPusher) push(
 		return fmt.Errorf("push: %w", err)
 	}
 	if res.Errors > 0 {
-		log.Printf(
-			"pg watch: pushed %d sessions, %d messages, %d errors (%s)",
-			res.SessionsPushed, res.MessagesPushed, res.Errors, reason,
-		)
+		logPGWatchPushResult(res, reason)
 		log.Printf(
 			"pg watch: %d session(s) failed to push; will retry",
 			res.Errors,
 		)
 		return nil
 	}
+	logPGWatchPushResult(res, reason)
+	return nil
+}
+
+func logPGWatchPushResult(res postgres.PushResult, reason pushReason) {
+	if res.SkippedConflicts > 0 {
+		log.Printf(
+			"pg watch: pushed %d sessions, %d messages, skipped %d ownership conflict(s), %d errors (%s)",
+			res.SessionsPushed, res.MessagesPushed,
+			res.SkippedConflicts, res.Errors, reason,
+		)
+		log.Printf(
+			"pg watch: %d session(s) skipped due to PostgreSQL ownership conflicts",
+			res.SkippedConflicts,
+		)
+		return
+	}
+	if res.Errors > 0 {
+		log.Printf(
+			"pg watch: pushed %d sessions, %d messages, %d errors (%s)",
+			res.SessionsPushed, res.MessagesPushed,
+			res.Errors, reason,
+		)
+		return
+	}
 	log.Printf(
 		"pg watch: pushed %d sessions, %d messages (%s)",
 		res.SessionsPushed, res.MessagesPushed, reason,
 	)
-	return nil
 }
 
 func (p *pgPusher) reset() {

cmd/agentsview/pg_watch_test.go

@@ -248,6 +248,35 @@ func TestPgPusher_LogsPartialPushErrors(t *testing.T) {
 	assert.Contains(t, got, "change")
 }
 
+func TestPgPusher_LogsSkippedConflicts(t *testing.T) {
+	target := &fakeTarget{
+		pushResult: postgres.PushResult{
+			SessionsPushed:   3,
+			MessagesPushed:   9,
+			SkippedConflicts: 2,
+		},
+	}
+	var logs bytes.Buffer
+	prev := log.Writer()
+	log.SetOutput(&logs)
+	t.Cleanup(func() { log.SetOutput(prev) })
+
+	p := &pgPusher{
+		localSync: func(context.Context) error { return nil },
+		connect: func() (pgTarget, error) {
+			return target, nil
+		},
+	}
+	require.NoError(t, p.push(context.Background(), reasonChange, false))
+
+	got := logs.String()
+	assert.Contains(t, got,
+		"pushed 3 sessions, 9 messages, skipped 2 ownership conflict(s), 0 errors")
+	assert.Contains(t, got,
+		"2 session(s) skipped due to PostgreSQL ownership conflicts")
+	assert.Contains(t, got, "change")
+}
+
 func TestResolveWatchTargets_ErrorsOnEmptyURL(t *testing.T) {
 	appCfg := config.Config{} // no PG URL
 	_, _, _, err := resolveWatchTargets(appCfg, PGPushConfig{})