for (int j = i + 1; j < rules.size(); j++) {
                assertNotSame(rules.get(i).getRuleId(), rules.get(j).getRuleId());
            }
        }
    }

    /**
     * Test rule processor chain behavior
     */
    public void test_ruleProcessorChain() {
        final List<String> executionOrder = new ArrayList<>();

        // Create custom processor that tracks execution

func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	c := make(chan ioret[V], 1)
	go func() {
		v, err := work(ctx)
		c <- ioret[V]{val: v, err: err}
	}()

	select {
	case v := <-c:
		return v.val, v.err
	case <-ctx.Done():
		var zero V

type LRU[K comparable, V any] struct {
	size      int
	evictList *LruList[K, V]
	items     map[K]*Entry[K, V]
	onEvict   EvictCallback[K, V]

	// expirable options
	mu   sync.Mutex
	ttl  time.Duration
	done chan struct{}

	// buckets for expiration
	buckets []bucket[K, V]
	// uint8 because it's number between 0 and numBuckets
	nextCleanupBucket uint8
}

// bucket is a container for holding entries to be expired

 *   <li>Supports a copy-free {@link #subArray} view, so methods that accept this type don't need to
 *       add overloads that accept start and end indexes.
 *   <li>Can be streamed without "breaking the chain": {@code foo.getBarLongs().stream()...}.
 *   <li>Access to all collection-based utilities via {@link #asList} (though at the cost of
 *       allocating garbage).
 * </ul>
 *
 * <p>Disadvantages compared to {@code long[]}:

 *   <li>Supports a copy-free {@link #subArray} view, so methods that accept this type don't need to
 *       add overloads that accept start and end indexes.
 *   <li>Can be streamed without "breaking the chain": {@code foo.getBarInts().stream()...}.
 *   <li>Access to all collection-based utilities via {@link #asList} (though at the cost of
 *       allocating garbage).
 * </ul>
 *
 * <p>Disadvantages compared to {@code int[]}:

        response = spy(response);
        when(response.getHeaderStart()).thenReturn(0);

        // Due to bug in line 96, data must be written at bufferIndex=8
        // Write entries with proper nextEntryOffset chain starting at 8
        int currentOffset = 8;
        for (int i = 0; i < 3; i++) {
            // Set NextEntryOffset (104 bytes to next, 0 for last)
            SMBUtil.writeInt4(i < 2 ? 104 : 0, buffer, currentOffset);

	case formatLZ4:
		r = lz4.NewReader(bf)
	case formatUnknown:
		r = bf
	default:
		return fmt.Errorf("Unsupported format %s", f)
	}
	tarReader := tar.NewReader(r)
	n := 0
	asyncWriters := make(chan struct{}, 16)
	var wg sync.WaitGroup

	var asyncErr error
	var asyncErrMu sync.Mutex
	for {
		if !o.ignoreErrs {
			asyncErrMu.Lock()
			err := asyncErr
			asyncErrMu.Unlock()
			if err != nil {

const maxDocumentSize = 10 << 20

// Reader - JSON record reader for S3Select.
type Reader struct {
	args       *ReaderArgs
	decoder    *jstream.Decoder
	valueCh    chan *jstream.MetaValue
	readCloser io.ReadCloser
}

// Read - reads single record.
func (r *Reader) Read(dst sql.Record) (sql.Record, error) {
	v, ok := <-r.valueCh
	if !ok {
		if err := r.decoder.Err(); err != nil {

	// expected in case of read timeout
	w.Header().Set("Connection", "close")

	setEventStreamHeaders(w)

	logCh := make(chan log.Info, 1000)
	peers, _ := newPeerRestClients(globalEndpoints)
	encodedCh := make(chan []byte, 1000+len(peers)*1000)
	err = globalConsoleSys.Subscribe(logCh, ctx.Done(), node, limitLines, logKind, nil)
	if err != nil {

// The returned function will return the results once there is enough or input is closed,
// or the context is canceled.
func (o *listPathOptions) gatherResults(ctx context.Context, in <-chan metaCacheEntry) func() (metaCacheEntriesSorted, error) {
	resultsDone := make(chan metaCacheEntriesSorted)
	// Copy so we can mutate
	resCh := resultsDone
	var done atomic.Bool
	resErr := io.EOF

	go func() {
		var results metaCacheEntriesSorted