outbound.go RawHistoryFolder

1package extensions
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"sync"
9	"time"
10)
11
12// orderedPipe serializes every host frame, not just notifications. Events
13// enqueue without waiting for the subprocess; synchronous requests wait for
14// their write. Overflow disconnects the consumer rather than silently dropping
15// an event and continuing with a misleading audit stream.
16type orderedPipe struct {
17	pipe   io.WriteCloser
18	mu     sync.Mutex
19	queue  []outboundFrame
20	bytes  int
21	closed bool
22	wake   chan struct{}
23	done   chan struct{}
24	once   sync.Once
25}
26
27type outboundFrame struct {
28	data []byte
29	ack  chan error
30}
31
32const outboundBytes = 16 * 1024 * 1024
33const outboundFrames = 256
34
35func newOrderedPipe(pipe io.WriteCloser) *orderedPipe {
36	p := &orderedPipe{pipe: pipe, wake: make(chan struct{}, 1), done: make(chan struct{})}
37	go p.run()
38	return p
39}
40
41func (p *orderedPipe) enqueue(data []byte, ack chan error) error {
42	p.mu.Lock()
43	if p.closed {
44		p.mu.Unlock()
45		return io.ErrClosedPipe
46	}
47	if len(p.queue) >= outboundFrames || p.bytes+len(data) > outboundBytes {
48		p.closed = true
49		p.mu.Unlock()
50		p.Close()
51		return errors.New("extension outbound queue exceeded its limit")
52	}
53	p.queue = append(p.queue, outboundFrame{data: append([]byte(nil), data...), ack: ack})
54	p.bytes += len(data)
55	p.mu.Unlock()
56	select {
57	case p.wake <- struct{}{}:
58	default:
59	}
60	return nil
61}
62
63func (p *orderedPipe) Write(data []byte) (int, error) {
64	return p.writeTimeout(data, interceptTimeout)
65}
66
67func (p *orderedPipe) writeTimeout(data []byte, timeout time.Duration) (int, error) {
68	if timeout <= 0 {
69		p.Close()
70		return 0, io.ErrClosedPipe
71	}
72	ack := make(chan error, 1)
73	if err := p.enqueue(data, ack); err != nil {
74		return 0, err
75	}
76	timer := time.NewTimer(timeout)
77	defer timer.Stop()
78	select {
79	case err := <-ack:
80		if err != nil {
81			return 0, err
82		}
83		return len(data), nil
84	case <-p.done:
85		return 0, io.ErrClosedPipe
86	case <-timer.C:
87		p.Close()
88		return 0, fmt.Errorf("extension outbound write timed out: %w", context.DeadlineExceeded)
89	}
90}
91
92func (p *orderedPipe) Close() error {
93	p.once.Do(func() {
94		p.mu.Lock()
95		p.closed = true
96		p.queue = nil
97		p.bytes = 0
98		p.mu.Unlock()
99		close(p.done)
100		_ = p.pipe.Close()
101	})
102	return nil
103}
104
105func (p *orderedPipe) run() {
106	for {
107		select {
108		case <-p.done:
109			return
110		case <-p.wake:
111		}
112		for {
113			p.mu.Lock()
114			if p.closed || len(p.queue) == 0 {
115				p.mu.Unlock()
116				break
117			}
118			frame := p.queue[0]
119			p.queue[0] = outboundFrame{}
120			p.queue = p.queue[1:]
121			p.bytes -= len(frame.data)
122			p.mu.Unlock()
123			n, err := p.pipe.Write(frame.data)
124			if err == nil && n != len(frame.data) {
125				err = io.ErrShortWrite
126			}
127			if frame.ack != nil {
128				frame.ack <- err
129			}
130			if err != nil {
131				p.Close()
132				return
133			}
134		}
135	}
136}