pico

created pr with 103.1 on 2026-01-05T14:52:02Z · by c8ef7d19

added 103.2 on 2026-01-08T01:19:34Z · by c8ef7d19


                      1:  66cf9b9 = 1:  66cf9b9 feat(pipe): add pipe_monitors table


                      2:  83b0c70 = 2:  83b0c70 chore(pipe): add new db methods


                      3:  43aa4a1 = 3:  43aa4a1 chore(pipe): add db impl


                      4:  a75186b = 4:  a75186b chore(pipe): add tests for monitoring


                      5:  9c7fcda = 5:  9c7fcda feat(pipe): add monitor cli


                      6:  2b86dec = 6:  2b86dec feat(pipe): status and rss commands


                      7:  8df2aa2 = 7:  8df2aa2 feat(pipe): monitor pub and pipe cmd


                      8:  11a5bce = 8:  11a5bce chore(pipe): monitor help text


                      9:  670350c = 9:  670350c refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 = 10:  ade2b83 fix: window and ping fixes


                      11:  68f5cd1 = 11:  68f5cd1 chore(pipe): add tests


                      -:  ------- > 12:  74fb40f chore(pipe): add logging stmts


                      -:  ------- > 13:  c422818 refactor(pipe): create pipe monitors history table


                      -:  ------- > 14:  e0c3a88 fix(pipe): only update monitor if within window


                      -:  ------- > 15:  9a8fbde chore(pipe): implement monitor history db interface


                      -:  ------- > 16:  f168507 feat(pipe): record historical monitors


                      -:  ------- > 17:  16df77f feat(pipe): monitor calculate uptime


                      -:  ------- > 18:  10ffdf1 feat(pipe): uptime cli command


                      -:  ------- > 19:  8231ac5 chore(pipe): monitor history stubs


                      -:  ------- > 20:  65a63a9 fix(pipe): uptime cmd

added 103.3 on 2026-02-01T17:16:24Z · by c8ef7d19


                      3:  43aa4a1 ! 1:  603ca6e chore(pubsub): add more tests


                      1:  66cf9b9 < -:  ------- feat(pipe): add pipe_monitors table


                      2:  83b0c70 < -:  ------- chore(pipe): add new db methods


                      4:  a75186b ! 2:  17e00b2 feat(pubsub): round robin


                      13:  c422818 ! 3:  e3136bd fix(pubsub): check for eof before processing and skip empty byte reads


                      -:  ------- > 4:  9a6d19e fix: rr


                      5:  9c7fcda < -:  ------- feat(pipe): add monitor cli


                      16:  f168507 ! 5:  5b3f3a1 fix: sending 0 byte read


                      6:  2b86dec < -:  ------- feat(pipe): status and rss commands


                      11:  68f5cd1 ! 6:  4fff471 refactor: fixes


                      7:  8df2aa2 < -:  ------- feat(pipe): monitor pub and pipe cmd


                      14:  e0c3a88 ! 7:  d0dfc85 chore: SetDispatch on Broker


                      8:  11a5bce < -:  ------- chore(pipe): monitor help text


                      9:  670350c < -:  ------- refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 < -:  ------- fix: window and ping fixes


                      12:  74fb40f < -:  ------- chore(pipe): add logging stmts


                      15:  9a8fbde < -:  ------- chore(pipe): implement monitor history db interface


                      17:  16df77f < -:  ------- feat(pipe): monitor calculate uptime


                      18:  10ffdf1 < -:  ------- feat(pipe): uptime cli command


                      19:  8231ac5 < -:  ------- chore(pipe): monitor history stubs


                      20:  65a63a9 < -:  ------- fix(pipe): uptime cmd

cmds

checkout latest patchset:

ssh pr.pico.sh print 103 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 103.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 103

Patchset `103.3` on 2026-02-01T17:16:24Z · commit 5b3f3a1

chore(pubsub): add more tests

Eric Bower 2026-01-25T16:38:25Z

feat(pubsub): round robin

Eric Bower 2026-01-25T16:38:25Z

fix(pubsub): check for eof before processing and skip empty byte reads

Eric Bower 2026-01-28T01:42:34Z

fix: rr

Eric Bower 2026-01-28T01:50:04Z

→ fix: sending 0 byte read

Eric Bower 2026-01-28T02:50:25Z

refactor: fixes

Eric Bower 2026-01-29T02:44:01Z

chore: SetDispatch on Broker

Eric Bower 2026-02-01T16:37:59Z

fix: sending 0 byte read

Eric Bower 2026-01-28T02:50:25Z

Semantic diff summary

0 added, 1 modified, 0 signature changed, 0 removed across 1 analyzed file

pkg/pubsub/broker.go

function_declaration Connect modified

+19 -11 pkg/pubsub/broker.go #

......@@ -105,20 +105,20 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
				data := make([]byte, 32*1024)
				n, err := client.ReadWriter.Read(data)

				// Check for EOF before processing
				if err != nil {
					if errors.Is(err, io.EOF) {
				// TODO: Skip empty reads
				/*
					if err != nil {
						if errors.Is(err, io.EOF) {
							return
						}
						inputErr = err
						return
					}
					inputErr = err
					return
				}

				// Skip empty reads
				if n == 0 {
					continue
				}

					if n == 0 {
						continue
					}
				*/
				data = data[:n]

				channelMessage := ChannelMessage{

......@@ -167,6 +167,14 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
				}

				sendwg.Wait()

				if err != nil {
					if errors.Is(err, io.EOF) {
						return
					}
					inputErr = err
					return
				}
			}
		}()
	}

...	...		@@ -105,20 +105,20 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
105	105		data := make([]byte, 32*1024)
106	106		n, err := client.ReadWriter.Read(data)
107	107
108		-	// Check for EOF before processing
109		-	if err != nil {
110		-	if errors.Is(err, io.EOF) {
	108	+	// TODO: Skip empty reads
	109	+	/*
	110	+	if err != nil {
	111	+	if errors.Is(err, io.EOF) {
	112	+	return
	113	+	}
	114	+	inputErr = err
111	115		return
112	116		}
113		-	inputErr = err
114		-	return
115		-	}
116		-
117		-	// Skip empty reads
118		-	if n == 0 {
119		-	continue
120		-	}
121	117
	118	+	if n == 0 {
	119	+	continue
	120	+	}
	121	+	*/
122	122		data = data[:n]
123	123
124	124		channelMessage := ChannelMessage{

...	...		@@ -167,6 +167,14 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
167	167		}
168	168
169	169		sendwg.Wait()
	170	+
	171	+	if err != nil {
	172	+	if errors.Is(err, io.EOF) {
	173	+	return
	174	+	}
	175	+	inputErr = err
	176	+	return
	177	+	}
170	178		}
171	179		}()
172	180		}

pico

Patchset 103.3 on 2026-02-01T17:16:24Z · commit 5b3f3a1

Patchset `103.3` on 2026-02-01T17:16:24Z · commit 5b3f3a1