Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
6.8 kB · 178 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179package oracle
// This file is compiled into the exact pinned upstream oracle package. It// reuses upstream's seed-derived chain, predicate kill selector, JSS observer,// ground-truth model, and post-restart assertion bundle while substituting a// real ReleaseSafe Stream process for upstream's Go daemon.
import ( "context" "fmt" "os/exec" "path/filepath" "strconv" "strings" "syscall" "testing" "time"
"github.com/stretchr/testify/require")
func TestStreamRestartChainBaselineOracle(t *testing.T) { cfg := Config{ Mode: "restart", Seed: restartSeed(0), Accounts: 4, MinInitialRecords: 1, MaxInitialRecords: 4, LiveEventsBootstrap: 4, LiveEventsSteady: 4, } spec := deriveChainSpec(cfg.Seed, cfg.Accounts) w := newRestartWorld(t, cfg) defer func() { require.NoError(t, w.Close()) }() coord := newChainCoordinator(t, w, spec) srv := newRestartServer(t, w, coord.onGetRepoServed) defer srv.Close() dataDir := filepath.Join(t.TempDir(), "stream") baseURL := "http://127.0.0.1:" + strconv.Itoa(freeStreamOraclePort(t)) logs := &synchronizedBuffer{} cmd := exec.Command(streamOracleBinary(t), streamRestartArgs(baseURL, dataDir, srv.URL, cfg.Accounts)...) cmd.Stdout = logs cmd.Stderr = logs require.NoError(t, cmd.Start()) waitForStreamOracleServing(t, cmd, baseURL, logs) watermark := streamCompactionWatermark(t, baseURL) require.NoError(t, cmd.Process.Signal(syscall.SIGTERM)) require.NoErrorf(t, waitStreamCommand(cmd, 15*time.Second), "baseline child did not stop cleanly:\n%s", logs.String()) assertOracleMatchesAfterReplay(t, dataDir, w, cfg, "stream-chain-baseline") cov := streamChainCoverage(t, dataDir, coord, watermark) assertStreamChainDurable(t, cov, coord, "stream-chain-baseline")}
func TestStreamRestartPredicateKillOracle(t *testing.T) { streamBin := streamOracleBinary(t) for i, seed := range predicateKillSeeds() { decision := selectKill(seed) t.Run(fmt.Sprintf("seed-%d-%s", seed, decision), func(t *testing.T) { cfg := Config{ Mode: "restart", Seed: restartSeed(i), Accounts: 4, MinInitialRecords: 1, MaxInitialRecords: 4, LiveEventsBootstrap: 4, LiveEventsSteady: 4, } spec := deriveChainSpec(cfg.Seed, cfg.Accounts) w := newRestartWorld(t, cfg) defer func() { require.NoError(t, w.Close()) }() coord := newChainCoordinator(t, w, spec) srv := newRestartServer(t, w, coord.onGetRepoServed) defer srv.Close()
dataDir := filepath.Join(t.TempDir(), "stream") baseURL := "http://127.0.0.1:" + strconv.Itoa(freeStreamOraclePort(t)) logs := &synchronizedBuffer{} args := streamRestartArgs(baseURL, dataDir, srv.URL, cfg.Accounts) crashSpec := fmt.Sprintf("%s:%d", decision.point, decision.ordinal) first := exec.Command(streamBin, append(args, "--crashpoint="+crashSpec)...) first.Stdout = logs first.Stderr = logs require.NoError(t, first.Start()) firstErr := waitStreamCommand(first, 60*time.Second) require.Error(t, firstErr, "armed Stream child exited cleanly") require.Contains(t, logs.String(), "crashpoint "+decision.point.String()+": aborting", "selected predicate kill never fired")
ctx, cancel := context.WithCancel(context.Background()) defer cancel() second := exec.CommandContext(ctx, streamBin, args...) second.Stdout = logs second.Stderr = logs require.NoError(t, second.Start()) waitForStreamOracleServing(t, second, baseURL, logs) watermark := streamCompactionWatermark(t, baseURL) require.NoError(t, second.Process.Signal(syscall.SIGTERM)) require.NoErrorf(t, waitStreamCommand(second, 15*time.Second), "recovery Stream child did not stop cleanly:\n%s", logs.String())
label := fmt.Sprintf("stream-predicate-%s-seed%d", decision, seed) assertOracleMatchesAfterReplay(t, dataDir, w, cfg, label) cov := streamChainCoverage(t, dataDir, coord, watermark) assertStreamChainDurable(t, cov, coord, label) rc := recordChainForShape(t, spec, shapeBfCreateDelete) assertCoverageFailsWithoutRow(t, cov, "delete", rc.collection, rc.rkey) require.NotContains(t, logs.String(), "fatal event handling failure") require.NotContains(t, logs.String(), "OutOfMemory") t.Logf("predicate receipt: seed=%d world_seed=%d point=%s ordinal=%d rows=%d", seed, cfg.Seed, decision.point, decision.ordinal, len(cov.events)) }) }}
func streamCompactionWatermark(t *testing.T, baseURL string) uint64 { t.Helper() metrics := scrapeStreamOracleMetrics(t, baseURL) value, ok := metricValue(metrics, "stream_compaction_watermark_seq") require.True(t, ok, "Stream compaction watermark metric is missing") require.GreaterOrEqual(t, value, float64(0)) return uint64(value)}
func streamChainCoverage(t *testing.T, dataDir string, coord *chainCoordinator, watermark uint64) chainCoverageView { t.Helper() events, err := ObserveSegments(dataDir) require.NoError(t, err, "observe Stream segments") events = EventsSortedBySeq(events) diskRows := NormalizeEventLog(events) wantPre := expectedChainRows(coord.hostDID, coord.recordedOps()) seqByKey := firstSeqByRowKey(diskRows) wantSeqed := make([]EventLogRow, len(wantPre)) for i, row := range wantPre { row.Seq = seqByKey[rowIdentity(row)] wantSeqed[i] = row } return chainCoverageView{ want: zeroRowSeqs(filterCompactedExpectedRowsWithObservedTombstones(wantSeqed, diskRows, watermark)), got: zeroRowSeqs(diskRows), events: events, watermark: watermark, }}
func assertStreamChainDurable(t *testing.T, cov chainCoverageView, coord *chainCoordinator, phase string) { t.Helper() require.NoErrorf(t, CompareEventLogCoverage(cov.want, cov.got), "%s: at-least-once event-log coverage (W=%d)", phase, cov.watermark) require.NoErrorf(t, CheckCompacted(cov.events, cov.watermark), "%s: compaction contract (W=%d)", phase, cov.watermark) truth, err := GroundTruthFromWorld(coord.w) require.NoErrorf(t, err, "%s: ground truth for visibility", phase) assertRecreatedRecordsVisible(t, cov.events, coord.spec, coord.hostDID, truth, phase)}
func streamRestartArgs(baseURL, dataDir, relayURL string, accounts int) []string { port := strings.TrimPrefix(baseURL, "http://127.0.0.1:") return []string{ "--port=" + port, "--data-dir=" + dataDir, "--relay-url=" + relayURL, "--plc-url=" + relayURL, "--max-backfill-repos=" + strconv.Itoa(accounts), "--backfill-workers=4", "--max-segment-bytes=1", "--compaction-interval=1s", "--retry-interval=0", }}
func waitStreamCommand(cmd *exec.Cmd, timeout time.Duration) error { done := make(chan error, 1) go func() { done <- cmd.Wait() }() select { case err := <-done: return err case <-time.After(timeout): _ = cmd.Process.Kill() <-done return fmt.Errorf("Stream child timed out after %s", timeout) }}