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	return p.writeContext(context.Background(), data, timeout)
69}
70
71func (p *orderedPipe) writeContext(ctx context.Context, data []byte, timeout time.Duration) (int, error) {
72	if err := ctx.Err(); err != nil {
73		return 0, err
74	}
75	if timeout <= 0 {
76		p.Close()
77		return 0, io.ErrClosedPipe
78	}
79	ack := make(chan error, 1)
80	if err := p.enqueue(data, ack); err != nil {
81		return 0, err
82	}
83	timer := time.NewTimer(timeout)
84	defer timer.Stop()
85	select {
86	case err := <-ack:
87		if err != nil {
88			return 0, err
89		}
90		return len(data), nil
91	case <-p.done:
92		return 0, io.ErrClosedPipe
93	case <-ctx.Done():
94		// A partially written frame cannot be safely withdrawn.
95		p.Close()
96		return 0, ctx.Err()
97	case <-timer.C:
98		p.Close()
99		return 0, fmt.Errorf("extension outbound write timed out: %w", context.DeadlineExceeded)
100	}
101}
102
103func (p *orderedPipe) Close() error {
104	p.once.Do(func() {
105		p.mu.Lock()
106		p.closed = true
107		p.queue = nil
108		p.bytes = 0
109		p.mu.Unlock()
110		close(p.done)
111		_ = p.pipe.Close()
112	})
113	return nil
114}
115
116func (p *orderedPipe) run() {
117	for {
118		select {
119		case <-p.done:
120			return
121		case <-p.wake:
122		}
123		for {
124			p.mu.Lock()
125			if p.closed || len(p.queue) == 0 {
126				p.mu.Unlock()
127				break
128			}
129			frame := p.queue[0]
130			p.queue[0] = outboundFrame{}
131			p.queue = p.queue[1:]
132			p.bytes -= len(frame.data)
133			p.mu.Unlock()
134			n, err := p.pipe.Write(frame.data)
135			if err == nil && n != len(frame.data) {
136				err = io.ErrShortWrite
137			}
138			if frame.ack != nil {
139				frame.ack <- err
140			}
141			if err != nil {
142				p.Close()
143				return
144			}
145		}
146	}
147}