1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
package handlers
// Test pinning the load-bearing call ordering in readFiles' output EOF
// epilogue: the handshake epoch (OutputEpoch) MUST be captured before the
// pending-work check (PendingAndActive). Joiners increment the pending count
// before enabling output mode, so capturing the epoch first guarantees that a
// joiner invisible to the pending==0 check bumps the epoch after the capture,
// turning the stale SignalOutputEOF into a no-op. Reordering the two calls
// would silently reopen the mid-batch EOF window; this test fails if someone
// does that.
import (
"context"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/mimecast/dtail/internal/lcontext"
"github.com/mimecast/dtail/internal/omode"
"github.com/mimecast/dtail/internal/regex"
)
// epochOrderTestServer wraps globCapTestServer and records the order of the
// output-handshake-relevant calls made by readFiles. It reports an enabled
// output session (DirectOutputActive true, HasOutputEOF
// true) so readFiles runs the full EOF epilogue.
type epochOrderTestServer struct {
*globCapTestServer
mu sync.Mutex
calls []string
signaledEpochs []uint64
outputLines chan []byte
}
func newEpochOrderTestServer() *epochOrderTestServer {
return &epochOrderTestServer{
globCapTestServer: newGlobCapTestServer(100),
outputLines: make(chan []byte, 100),
}
}
func (s *epochOrderTestServer) record(name string) {
s.mu.Lock()
defer s.mu.Unlock()
s.calls = append(s.calls, name)
}
func (s *epochOrderTestServer) recordedCalls() []string {
s.mu.Lock()
defer s.mu.Unlock()
return append([]string(nil), s.calls...)
}
func (s *epochOrderTestServer) DirectOutputActive() bool { return true }
func (s *epochOrderTestServer) EnableDirectOutput() bool { return false }
func (s *epochOrderTestServer) HasOutputEOF() bool { return true }
func (s *epochOrderTestServer) GetOutputChannel() chan []byte { return s.outputLines }
// OutputEpoch returns a sentinel value so the test can also verify that the
// captured epoch is the one passed through to SignalOutputEOF.
func (s *epochOrderTestServer) OutputEpoch() uint64 {
s.record("OutputEpoch")
return 42
}
func (s *epochOrderTestServer) PendingAndActive() (int32, int32) {
s.record("PendingAndActive")
return s.globCapTestServer.PendingAndActive()
}
func (s *epochOrderTestServer) FlushOutput() {
s.record("FlushOutput")
}
func (s *epochOrderTestServer) SignalOutputEOF(epoch uint64) {
s.mu.Lock()
s.signaledEpochs = append(s.signaledEpochs, epoch)
s.mu.Unlock()
s.record("SignalOutputEOF")
}
var _ readCommandServer = (*epochOrderTestServer)(nil)
// TestReadFilesCapturesEpochBeforePendingCheck drives readFiles over a real
// (empty) file with an enabled output session and asserts that the EOF
// epilogue runs exactly OutputEpoch -> PendingAndActive -> FlushOutput ->
// SignalOutputEOF, i.e. the epoch is captured before the pending check.
// Earlier PendingAndActive calls from the per-file phase are recorded too,
// which is why the assertion checks the trailing four calls.
func TestReadFilesCapturesEpochBeforePendingCheck(t *testing.T) {
resetServerLogger(t)
dir := t.TempDir()
path := filepath.Join(dir, "empty.log")
if err := os.WriteFile(path, nil, 0o600); err != nil {
t.Fatalf("create temp file: %v", err)
}
srv := newEpochOrderTestServer()
cmd := newReadCommand(srv, omode.CatClient)
cmd.readFiles(context.Background(), lcontext.LContext{}, []string{path}, path,
regex.NewNoop(), time.Millisecond)
calls := srv.recordedCalls()
wantTail := []string{"OutputEpoch", "PendingAndActive", "FlushOutput", "SignalOutputEOF"}
if len(calls) < len(wantTail) {
t.Fatalf("EOF epilogue did not run, recorded calls: %v", calls)
}
tail := calls[len(calls)-len(wantTail):]
for i, want := range wantTail {
if tail[i] != want {
t.Fatalf("EOF epilogue call order = %v, want trailing %v (epoch must be captured BEFORE the pending check)",
tail, wantTail)
}
}
srv.mu.Lock()
signaled := append([]uint64(nil), srv.signaledEpochs...)
srv.mu.Unlock()
if len(signaled) != 1 || signaled[0] != 42 {
t.Fatalf("SignalOutputEOF epochs = %v, want exactly [42] (the captured OutputEpoch value)", signaled)
}
}
|
