test(http): allow concurrent early match and exercise redirect input

go.mod

@@ -102,7 +102,7 @@ require (
 	github.com/projectdiscovery/tlsx v1.4.3-0.20261020052910-acb80c2c7c75
 	github.com/projectdiscovery/uncover v1.2.1
 	github.com/projectdiscovery/useragent v0.0.109
-	github.com/projectdiscovery/utils v0.11.10
+	github.com/projectdiscovery/utils v0.11.11-0.20261020211236-cde16c1381ec
 	github.com/projectdiscovery/wappalyzergo v0.3.5
 	github.com/projectdiscovery/yamldoc-go v1.0.7
 	github.com/redis/go-redis/v9 v9.19.0

go.sum

@@ -999,8 +999,8 @@ github.com/projectdiscovery/uncover v1.2.1 h1:8U46T/96CLT7BPoXBgkTvWqB06lOyeTSLv
 github.com/projectdiscovery/uncover v1.2.1/go.mod h1:0p8onrWxfpXQEYs90ZDzTSpu1107gWmodX1NWqu/+z4=
 github.com/projectdiscovery/useragent v0.0.109 h1:b86BPdZvmgbJousa04jAugDRY4Y+rAvEamE0TFCPmeE=
 github.com/projectdiscovery/useragent v0.0.109/go.mod h1:sRyBqDAcD31RKYrO8fyXGowLd36fNwY1g0MNW9mp7zk=
-github.com/projectdiscovery/utils v0.11.10 h1:6mtjWZifPT+UFXCeWSSzwb10KadrlkhYTm03jfhJ2EY=
-github.com/projectdiscovery/utils v0.11.10/go.mod h1:T7KzyElivWJiPP+B4sXqkp1CcFe6O/P0qBn/MjDNa74=
+github.com/projectdiscovery/utils v0.11.11-0.20261020211236-cde16c1381ec h1:VhJiivJR8ljv5Qk3QZQC9og50wspEBB5qkNfdyqNhws=
+github.com/projectdiscovery/utils v0.11.11-0.20261020211236-cde16c1381ec/go.mod h1:T7KzyElivWJiPP+B4sXqkp1CcFe6O/P0qBn/MjDNa74=
 github.com/projectdiscovery/wappalyzergo v0.3.5 h1:UZradYF1GIcruSIV5wwo/Coph9/VnM17oajLD9eXR2U=
 github.com/projectdiscovery/wappalyzergo v0.3.5/go.mod h1:4E6+fh/ckaY3wo5tlM7lNYSkhuTYRHzu2YmGIRraXP0=
 github.com/projectdiscovery/yamldoc-go v1.0.7 h1:Odmjd1o4EkMff2F6QHi1dIYPIQPZUFy24YqmQCprmTk=

internal/tests/integration/multi_test.go

@@ -17,7 +17,7 @@ type multiProtoDynamicExtractor struct{}
 
 // Execute executes a test case and returns an error if occurred
 func (h *multiProtoDynamicExtractor) Execute(templatePath string) error {
-	results, err := testutils.RunNucleiTemplateAndGetResults(templatePath, "docs.projectdiscovery.io", debug)
+	results, err := testutils.RunNucleiTemplateAndGetResults(templatePath, "https://docs.projectdiscovery.io/", debug)
 	if err != nil {
 		return err
 	}

pkg/protocols/common/contextargs/contextargs.go

@@ -256,6 +256,12 @@ func (ctx *Context) Clone() *Context {
 	return newCtx
 }
 
+func (ctx *Context) CloneWithContext(parent context.Context) *Context {
+	cloned := ctx.Clone()
+	cloned.ctx = parent
+	return cloned
+}
+
 // resolvePortList converts a list of port strings (numeric or service names) to numeric port strings.
 func resolvePortList(ports []string) []string {
 	resolved := make([]string, 0, len(ports))

pkg/protocols/common/contextargs/contextargs_test.go

@@ -7,6 +7,25 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestCloneWithContextPreservesInputWithoutSharingMutableState(t *testing.T) {
+	parent := NewWithInput(context.Background(), "https://example.com")
+	parent.Set("shared", "original")
+	childCtx, cancel := context.WithCancel(context.Background())
+	child := parent.CloneWithContext(childCtx)
+
+	require.Equal(t, parent.MetaInput.Input, child.MetaInput.Input)
+	childValue, ok := child.Get("shared")
+	require.True(t, ok)
+	require.Equal(t, "original", childValue)
+	child.Set("shared", "changed")
+	parentValue, ok := parent.Get("shared")
+	require.True(t, ok)
+	require.Equal(t, "original", parentValue)
+	cancel()
+	require.ErrorIs(t, child.Context().Err(), context.Canceled)
+	require.NoError(t, parent.Context().Err())
+}
+
 // TestUseNetworkPort is the behavior matrix for UseNetworkPort.
 //
 // The contract (see issue #7323):

pkg/protocols/common/protocolstate/memguardian.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/docker/go-units"
 	"github.com/projectdiscovery/gologger"
 	"github.com/projectdiscovery/utils/env"
 	httputil "github.com/projectdiscovery/utils/http"
@@ -16,6 +17,7 @@ var (
 	MaxBytesBufferAllocOnLowMemory = env.GetEnvOrDefault("MEMGUARDIAN_ALLOC", 0)
 	memTimer                       *time.Ticker
 	cancelFunc                     context.CancelFunc
+	memWatcherDone                 chan struct{}
 	muGlobalChange                 sync.Mutex
 )
 
@@ -25,12 +27,18 @@ func StartActiveMemGuardian(ctx context.Context) {
 	if memguardian.DefaultMemGuardian == nil || memTimer != nil {
 		return
 	}
+	if err := refreshMemoryWarning(); err != nil {
+		gologger.Warning().Msgf("could not check memory pressure: %s", err)
+	}
 
 	memTimer = time.NewTicker(memguardian.DefaultInterval)
 	ctx, cancelFunc = context.WithCancel(ctx)
+	memWatcherDone = make(chan struct{})
 
 	ticker := memTimer
+	done := memWatcherDone
 	go func(t *time.Ticker) {
+		defer close(done)
 		if t == nil {
 			return
 		}
@@ -39,6 +47,9 @@ func StartActiveMemGuardian(ctx context.Context) {
 			case <-ctx.Done():
 				return
 			case <-t.C:
+				if err := refreshMemoryWarning(); err != nil {
+					gologger.Warning().Msgf("could not check memory pressure: %s", err)
+				}
 				if IsLowOnMemory() {
 					_ = GlobalGuardBytesBufferAlloc()
 				} else {
@@ -49,6 +60,24 @@ func StartActiveMemGuardian(ctx context.Context) {
 	}(ticker)
 }
 
+func refreshMemoryWarning() error {
+	usedRatio, usedBytes, err := memguardian.UsedRam()
+	if err != nil {
+		return err
+	}
+
+	warning := usedRatio >= memguardian.DefaultMaxUsedRamRatio
+	if maxRam := env.GetEnvOrDefault(memguardian.MemGuardianMaxUsedMemoryENV, ""); maxRam != "" {
+		maxBytes, err := units.FromHumanSize(maxRam)
+		if err != nil {
+			return err
+		}
+		warning = warning || maxBytes > 0 && usedBytes >= uint64(maxBytes)
+	}
+	memguardian.DefaultMemGuardian.Warning.Store(warning)
+	return nil
+}
+
 func StopActiveMemGuardian() {
 	muGlobalChange.Lock()
 	defer muGlobalChange.Unlock()
@@ -61,6 +90,10 @@ func StopActiveMemGuardian() {
 		cancelFunc()
 		cancelFunc = nil
 	}
+	if memWatcherDone != nil {
+		<-memWatcherDone
+		memWatcherDone = nil
+	}
 	if memTimer != nil {
 		memTimer.Stop()
 		memTimer = nil
@@ -77,7 +110,7 @@ func IsLowOnMemory() bool {
 // GuardThreads on caller
 func GuardThreadsOrDefault(current int) int {
 	if MaxThreadsOnLowMemory > 0 {
-		return MaxThreadsOnLowMemory
+		return min(current, MaxThreadsOnLowMemory)
 	}
 
 	fraction := int(current / 5)

pkg/protocols/common/protocolstate/memguardian_test.go

@@ -127,3 +127,45 @@ func TestMemGuardianReset(t *testing.T) {
 	StopActiveMemGuardian()
 	time.Sleep(10 * time.Millisecond) // Allow cleanup
 }
+
+func TestActiveMemGuardianUpdatesLowMemoryWarning(t *testing.T) {
+	StopActiveMemGuardian()
+	t.Setenv(memguardian.MemGuardianMaxUsedMemoryENV, "")
+	original := memguardian.DefaultMemGuardian
+	originalInterval := memguardian.DefaultInterval
+	originalRatio := memguardian.DefaultMaxUsedRamRatio
+	memguardian.DefaultInterval = 5 * time.Millisecond
+	memguardian.DefaultMaxUsedRamRatio = 1
+	guardian, err := memguardian.New(
+		memguardian.WitInterval(memguardian.DefaultInterval),
+		memguardian.WithMaxRamRatioWarning(1),
+	)
+	require.NoError(t, err)
+	memguardian.DefaultMemGuardian = guardian
+	t.Cleanup(func() {
+		StopActiveMemGuardian()
+		guardian.Close()
+		memguardian.DefaultMemGuardian = original
+		memguardian.DefaultInterval = originalInterval
+		memguardian.DefaultMaxUsedRamRatio = originalRatio
+	})
+
+	StartActiveMemGuardian(context.Background())
+	require.Eventually(t, IsLowOnMemory, 100*time.Millisecond, 5*time.Millisecond)
+	StopActiveMemGuardian()
+	memguardian.DefaultMaxUsedRamRatio = 100
+	require.NoError(t, refreshMemoryWarning())
+	require.False(t, IsLowOnMemory())
+	t.Setenv(memguardian.MemGuardianMaxUsedMemoryENV, "0")
+	require.NoError(t, refreshMemoryWarning())
+	require.False(t, IsLowOnMemory())
+}
+
+func TestGuardThreadsNeverIncreasesConcurrency(t *testing.T) {
+	previous := MaxThreadsOnLowMemory
+	MaxThreadsOnLowMemory = 5
+	t.Cleanup(func() { MaxThreadsOnLowMemory = previous })
+
+	require.Equal(t, 2, GuardThreadsOrDefault(2))
+	require.Equal(t, 5, GuardThreadsOrDefault(25))
+}

pkg/protocols/headless/engine/page_actions.go

@@ -230,7 +230,9 @@ func (p *Page) WaitVisible(act *Action, out ActionData) error {
 }
 
 func (p *Page) Sleeper(pollTimeout, timeout time.Duration) *Page {
+	p.mutex.RLock()
 	page := *p
+	p.mutex.RUnlock()
 	page.page = page.Page().Sleeper(func() utils.Sleeper {
 		return createBackOffSleeper(pollTimeout, timeout)
 	})
@@ -238,7 +240,9 @@ func (p *Page) Sleeper(pollTimeout, timeout time.Duration) *Page {
 }
 
 func (p *Page) Timeout(timeout time.Duration) *Page {
+	p.mutex.RLock()
 	page := *p
+	p.mutex.RUnlock()
 	page.page = page.Page().Timeout(timeout)
 	return &page
 }
@@ -732,6 +736,18 @@ func (p *Page) SelectInputElement(act *Action, out ActionData) error {
 
 // WaitPageLifecycleEvent waits for specified page lifecycle event name
 func (p *Page) WaitPageLifecycleEvent(act *Action, out ActionData, event proto.PageLifecycleEventName) error {
+	if event == proto.PageLifecycleEventNameLoad {
+		dur, err := getTimeout(p, act)
+		if err != nil {
+			return err
+		}
+		if err := p.page.Timeout(dur).WaitLoad(); err != nil {
+			return err
+		}
+		p.updateLastNavigatedURL()
+		return nil
+	}
+
 	fn, err := getNavigationFunc(p, act, event)
 	if err != nil {
 		return err

pkg/protocols/headless/engine/page_actions_test.go

@@ -18,6 +18,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-rod/rod"
+	"github.com/go-rod/rod/lib/proto"
 	"github.com/stretchr/testify/require"
 
 	"github.com/projectdiscovery/nuclei/v3/internal/tests/testheadless"
@@ -153,6 +155,26 @@ func TestLocatorInteractshTracking(t *testing.T) {
 	})
 }
 
+func TestSleeperConcurrentHistoryUpdate(t *testing.T) {
+	page := &Page{page: &rod.Page{}, mutex: &sync.RWMutex{}}
+	start := make(chan struct{})
+	var done sync.WaitGroup
+	done.Add(1)
+	go func() {
+		defer done.Done()
+		<-start
+		for range 10000 {
+			page.addToHistory(HistoryData{})
+		}
+	}()
+
+	close(start)
+	for range 10000 {
+		page.Sleeper(time.Millisecond, time.Second)
+	}
+	done.Wait()
+}
+
 func TestActionNavigate(t *testing.T) {
 	response := `
 		<html>
@@ -642,6 +664,47 @@ func TestActionWaitLoad(t *testing.T) {
 	})
 }
 
+func TestActionWaitLoadAfterPageAlreadyLoaded(t *testing.T) {
+	actions := []*Action{{ActionType: ActionTypeHolder{ActionType: ActionNavigate}, Data: map[string]string{"url": "{{BaseURL}}"}}}
+	testHeadlessSimpleResponse(t, "<html><body>ready</body></html>", actions, 10*time.Second, func(page *Page, err error, _ ActionData) {
+		require.NoError(t, err)
+		require.NoError(t, page.Page().WaitLoad())
+
+		wait := &Action{ActionType: ActionTypeHolder{ActionType: ActionWaitLoad}, Data: map[string]string{"timeout": "2s"}}
+		started := time.Now()
+		require.NoError(t, page.WaitPageLifecycleEvent(wait, nil, proto.PageLifecycleEventNameLoad))
+		require.Less(t, time.Since(started), 1500*time.Millisecond, "a completed load must not consume the full wait timeout")
+	})
+}
+
+func TestActionWaitLoadWaitsForPendingResource(t *testing.T) {
+	resourceDone := make(chan struct{})
+	actions := []*Action{
+		{ActionType: ActionTypeHolder{ActionType: ActionNavigate}, Data: map[string]string{"url": "{{BaseURL}}"}},
+		{ActionType: ActionTypeHolder{ActionType: ActionWaitLoad}},
+	}
+	testHeadless(t, actions, 10*time.Second, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/slow.js" {
+			w.Header().Set("Content-Type", "text/javascript")
+			time.Sleep(300 * time.Millisecond)
+			_, _ = fmt.Fprint(w, "window.resourceLoaded = true")
+			close(resourceDone)
+			return
+		}
+		_, _ = fmt.Fprint(w, `<html><script src="/slow.js"></script></html>`)
+	}, func(page *Page, err error, _ ActionData) {
+		require.NoError(t, err)
+		select {
+		case <-resourceDone:
+		default:
+			t.Fatal("waitload returned before the pending resource completed")
+		}
+		loaded, evalErr := page.Page().Eval(`() => window.resourceLoaded`)
+		require.NoError(t, evalErr)
+		require.True(t, loaded.Value.Bool())
+	})
+}
+
 func TestActionGetResource(t *testing.T) {
 	response := `
 		<html>

pkg/protocols/http/http.go

@@ -21,7 +21,6 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/expressions"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
-	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/protocolstate"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/http/httpclientpool"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/network/networkclientpool"
 	"github.com/projectdiscovery/nuclei/v3/pkg/types/scanstrategy"
@@ -123,7 +122,8 @@ type Request struct {
 	// examples:
 	//   - name: Send requests using 10 concurrent threads
 	//     value: 10
-	Threads int `yaml:"threads,omitempty" json:"threads,omitempty" jsonschema:"title=threads for sending requests,description=Threads specifies number of threads to use sending requests. This enables Connection Pooling"`
+	Threads        int `yaml:"threads,omitempty" json:"threads,omitempty" jsonschema:"title=threads for sending requests,description=Threads specifies number of threads to use sending requests. This enables Connection Pooling"`
+	PayloadThreads int `yaml:"payload-threads,omitempty" json:"payload-threads,omitempty" jsonschema:"title=dependent payload concurrency,description=Opt-in concurrency across independent payload groups while preserving request order inside each group"`
 	// description: |
 	//   MaxSize is the maximum size of http response body to read in bytes.
 	// examples:
@@ -139,14 +139,15 @@ type Request struct {
 
 	CompiledOperators *operators.Operators `yaml:"-" json:"-"`
 
-	options           *protocols.ExecutorOptions
-	connConfiguration *httpclientpool.Configuration
-	totalRequests     int
-	customHeaders     map[string]string
-	generator         *generators.PayloadGenerator // optional, only enabled when using payloads
-	rawhttpClient     *rawhttp.Client
-	replayClient      *http.Client // set when -replay-proxy is given
-	dialer            *fastdialer.Dialer
+	options                 *protocols.ExecutorOptions
+	connConfiguration       *httpclientpool.Configuration
+	totalRequests           int
+	customHeaders           map[string]string
+	generator               *generators.PayloadGenerator // optional, only enabled when using payloads
+	dependentPayloadThreads int
+	rawhttpClient           *rawhttp.Client
+	replayClient            *http.Client // set when -replay-proxy is given
+	dialer                  *fastdialer.Dialer
 
 	// description: |
 	//   SelfContained specifies if the request is self-contained.
@@ -521,6 +522,7 @@ func (request *Request) Compile(options *protocols.ExecutorOptions) error {
 		delete(request.Payloads, payload)
 	}
 
+	request.dependentPayloadThreads = 0
 	if len(request.Payloads) > 0 {
 		request.generator, err = generators.New(request.Payloads, request.AttackType.Value, request.options.TemplatePath, request.options.Catalog, request.options.Options.AttackType, request.options.Options)
 		if err != nil {
@@ -547,12 +549,8 @@ func (request *Request) Compile(options *protocols.ExecutorOptions) error {
 		}
 	}
 	if len(request.Payloads) > 0 {
-		// Due to a known issue (https://github.com/projectdiscovery/nuclei/issues/5015),
-		// dynamic extractors cannot be used with payloads. To address this,
-		// execution is handled by the standard engine without concurrency,
-		// achieved by setting the thread count to 0.
-
-		// this limitation will be removed once we have a better way to handle dynamic extractors with payloads
+		// Multi-request matchers and dynamic extractors depend on ordered responses
+		// across a payload iteration, so they use the standard engine.
 		hasMultipleRequests := false
 		if len(request.Raw)+len(request.Path) > 1 {
 			hasMultipleRequests = true
@@ -565,17 +563,21 @@ func (request *Request) Compile(options *protocols.ExecutorOptions) error {
 				break
 			}
 		}
-		if hasNamedInternalExtractor && hasMultipleRequests {
-			stats.Increment(SetThreadToCountZero)
+		if hasMultipleRequests && (hasNamedInternalExtractor || request.NeedsRequestCondition()) {
+			if !hasNamedInternalExtractor && request.NeedsRequestCondition() && request.PayloadThreads > 0 && request.canRunDependentPayloadsConcurrently() {
+				request.dependentPayloadThreads = request.PayloadThreads
+			}
+			if request.dependentPayloadThreads == 0 {
+				stats.Increment(SetThreadToCountZero)
+			}
 			request.Threads = 0
 		} else {
-			// specifically for http requests high concurrency and threads will lead to memory exhaustion, hence reduce the maximum parallelism
-			if protocolstate.IsLowOnMemory() {
-				request.Threads = protocolstate.GuardThreadsOrDefault(request.Threads)
-			}
 			request.Threads = options.GetThreadsForNPayloadRequests(request.Requests(), request.Threads)
 		}
 	}
+	if request.PayloadThreads > 0 && request.dependentPayloadThreads == 0 {
+		return errors.New("payload-threads requires multiple raw requests, indexed response conditions, and independent payload groups without iterate-all, interactsh, or @once")
+	}
 
 	// Avoid reusing client-side HTTP proxy connections for the legacy non-threaded spray path.
 	if shouldDisableKeepAliveForHTTPProxy(request, options) {
@@ -622,7 +624,7 @@ const (
 )
 
 func init() {
-	stats.NewEntry(SetThreadToCountZero, "Setting thread count to 0 for %d templates, dynamic extractors are not supported with payloads yet")
+	stats.NewEntry(SetThreadToCountZero, "Setting thread count to 0 for %d templates with payloads and dependent requests")
 }
 
 // UpdateOptions replaces this request's options with a new copy
@@ -641,7 +643,7 @@ func shouldDisableKeepAliveForHTTPProxy(request *Request, options *protocols.Exe
 	if request == nil || options == nil || options.Options == nil || options.Options.AliveHttpProxy == "" {
 		return false
 	}
-	return request.Threads <= 0 && options.Options.ScanStrategy != scanstrategy.HostSpray.String()
+	return request.Threads <= 0 && request.dependentPayloadThreads <= 0 && options.Options.ScanStrategy != scanstrategy.HostSpray.String()
 }
 
 // AnalyzeConnectionReuse determines if a request can safely reuse connections.

pkg/protocols/http/httputils/spm.go

@@ -58,7 +58,7 @@ func NewBlockingSPMHandler[T comparable](ctx context.Context, size int, maxResul
 		maxResults:  maxResults,
 	}
 	s.internalWg.Add(1)
-	go s.run(ctx)
+	go s.run(ctx1)
 	return s
 }
 
@@ -77,7 +77,7 @@ func NewNonBlockingSPMHandler[T comparable](ctx context.Context, maxResults int,
 		maxResults:  maxResults,
 	}
 	s.internalWg.Add(1)
-	go s.run(ctx)
+	go s.run(ctx1)
 	return s
 }
 
@@ -132,18 +132,20 @@ func (h *StopAtFirstMatchHandler[T]) run(ctx context.Context) {
 			if !ok {
 				return
 			}
-			if h.onResult != nil {
-				h.onResult(val)
-			}
-			if len(h.results) >= h.maxResults {
-				// skip or do not store the result
-				continue
-			}
-			h.results[val] = struct{}{}
+			h.recordResult(val)
 		}
 	}
 }
 
+func (h *StopAtFirstMatchHandler[T]) recordResult(val T) {
+	if h.onResult != nil {
+		h.onResult(val)
+	}
+	if len(h.results) < h.maxResults {
+		h.results[val] = struct{}{}
+	}
+}
+
 // Context is canceled when the handler stops.
 func (h *StopAtFirstMatchHandler[T]) Context() context.Context {
 	return h.ctx
@@ -210,6 +212,9 @@ func (h *StopAtFirstMatchHandler[T]) Wait() {
 	// after waiting it closes the error channel
 	close(h.ResultChan)
 	h.internalWg.Wait()
+	for val := range h.ResultChan {
+		h.recordResult(val)
+	}
 }
 
 // CombinedResults returns the combined results

pkg/protocols/http/httputils/spm_test.go

@@ -0,0 +1,79 @@
+package httputils
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestCancelStopsResultReceiver(t *testing.T) {
+	constructors := map[string]func(context.Context) *StopAtFirstMatchHandler[int]{
+		"blocking": func(ctx context.Context) *StopAtFirstMatchHandler[int] {
+			return NewBlockingSPMHandler[int](ctx, 1, 1, true)
+		},
+		"nonblocking": func(ctx context.Context) *StopAtFirstMatchHandler[int] {
+			return NewNonBlockingSPMHandler[int](ctx, 1, true)
+		},
+	}
+
+	for name, construct := range constructors {
+		t.Run(name, func(t *testing.T) {
+			parent, cancelParent := context.WithCancel(context.Background())
+			defer cancelParent()
+			handler := construct(parent)
+			done := make(chan struct{})
+			go func() {
+				handler.internalWg.Wait()
+				close(done)
+			}()
+
+			handler.Cancel()
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+				t.Fatal("result receiver remained alive after cancellation")
+			}
+		})
+	}
+}
+
+func TestCancelPreservesQueuedResults(t *testing.T) {
+	constructors := map[string]func() *StopAtFirstMatchHandler[int]{
+		"blocking": func() *StopAtFirstMatchHandler[int] {
+			return NewBlockingSPMHandler[int](context.Background(), 1, 2, false)
+		},
+		"nonblocking": func() *StopAtFirstMatchHandler[int] {
+			return NewNonBlockingSPMHandler[int](context.Background(), 2, false)
+		},
+	}
+
+	for name, construct := range constructors {
+		t.Run(name, func(t *testing.T) {
+			for range 25 {
+				handler := construct()
+				entered := make(chan struct{})
+				release := make(chan struct{})
+				handler.SetOnResultCallback(func(value int) {
+					if value == 1 {
+						close(entered)
+						<-release
+					}
+				})
+				handler.ResultChan <- 1
+				<-entered
+				handler.ResultChan <- 2
+				handler.Cancel()
+				close(release)
+				handler.Wait()
+
+				found := make(map[int]bool)
+				for _, result := range handler.CombinedResults() {
+					found[result] = true
+				}
+				if !found[1] || !found[2] {
+					t.Fatalf("queued results lost after cancellation: %v", found)
+				}
+			}
+		})
+	}
+}

pkg/protocols/http/proxy_keepalive_test.go

@@ -13,6 +13,8 @@ func TestShouldDisableKeepAliveForHTTPProxy(t *testing.T) {
 	opts.AliveHttpProxy = "http://127.0.0.1:8080"
 	opts.ScanStrategy = scanstrategy.TemplateSpray.String()
 	require.True(t, shouldDisableKeepAliveForHTTPProxy(&Request{}, &protocols.ExecutorOptions{Options: opts}))
+	require.False(t, shouldDisableKeepAliveForHTTPProxy(&Request{Threads: 10}, &protocols.ExecutorOptions{Options: opts}))
+	require.False(t, shouldDisableKeepAliveForHTTPProxy(&Request{dependentPayloadThreads: 2}, &protocols.ExecutorOptions{Options: opts}))
 
 	opts.ScanStrategy = scanstrategy.HostSpray.String()
 	require.False(t, shouldDisableKeepAliveForHTTPProxy(&Request{}, &protocols.ExecutorOptions{Options: opts}))

pkg/protocols/http/request.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/pkg/errors"
 	"go.uber.org/multierr"
+	"golang.org/x/sync/semaphore"
 	"moul.io/http2curl"
 
 	"github.com/projectdiscovery/fastdialer/fastdialer"
@@ -122,13 +123,26 @@ func (request *Request) waitHostBackoff(ctx context.Context, host string) error
 	return request.options.HostBackoffWait(ctx, host)
 }
 
+type httpRateLimitWaitObserver interface {
+	ObserveHTTPRateLimitWait(time.Duration)
+}
+
+type httpRequestTimingObserver interface {
+	ObserveHTTPRequest(time.Duration, int, error)
+}
+
 // rateLimitTake first honours any backoff the host earned by signalling it is
 // overloaded, so a blocking target is paced regardless of which limiter is in play.
 func (request *Request) rateLimitTake(ctx context.Context, hostname string) error {
 	if err := request.waitHostBackoff(ctx, hostname); err != nil {
 		return err
 	}
-	return request.options.RateLimitTakeContext(ctx, hostname)
+	started := time.Now()
+	err := request.options.RateLimitTakeContext(ctx, hostname)
+	if observer, ok := request.options.Progress.(httpRateLimitWaitObserver); ok {
+		observer.ObserveHTTPRateLimitWait(time.Since(started))
+	}
+	return err
 }
 
 // stopWithHandler ends when parent ends or the handler stops, so a backoff
@@ -419,8 +433,12 @@ func (request *Request) executeParallelHTTP(input *contextargs.Context, dynamicV
 		}
 
 		// resize check point - nop if there are no changes
-		if shouldFollowGlobal && spmHandler.Size() != request.options.Options.PayloadConcurrency {
-			if err := spmHandler.Resize(input.Context(), request.options.Options.PayloadConcurrency); err != nil {
+		desiredWorkers := request.options.Options.PayloadConcurrency
+		if protocolstate.IsLowOnMemory() {
+			desiredWorkers = protocolstate.GuardThreadsOrDefault(desiredWorkers)
+		}
+		if shouldFollowGlobal && spmHandler.Size() != desiredWorkers {
+			if err := spmHandler.Resize(input.Context(), desiredWorkers); err != nil {
 				close(tasks)
 				workersWg.Wait()
 				return err
@@ -644,29 +662,62 @@ func (request *Request) ExecuteWithResults(input *contextargs.Context, dynamicVa
 	if request.Threads > 0 && (len(request.Payloads) > 0 || request.Race) {
 		return request.executeParallelHTTP(input, dynamicValues, callback)
 	}
+	if request.dependentPayloadThreads > 0 {
+		return request.executeDependentPayloads(input, dynamicValues, previous, callback)
+	}
+	return request.executeSequentialHTTP(input, dynamicValues, previous, callback, request.newGenerator(false), nil, nil)
+}
 
-	generator := request.newGenerator(false)
+func (request *Request) executeSequentialHTTP(input *contextargs.Context, dynamicValues, previous output.InternalEvent, callback protocols.OutputEventCallback, generator *requestGenerator, accounted *atomic.Int64, stopOnMatch *atomic.Bool) error {
+	var recordedCurrent bool
+	countAccounted := func(count int64) {
+		if accounted != nil {
+			accounted.Add(count)
+		}
+	}
+	recordSkipped := func(count int64) {
+		countAccounted(count)
+		request.options.Progress.SetRequests(uint64(count))
+	}
+	recordSent := func() {
+		recordedCurrent = true
+		countAccounted(1)
+		request.options.Progress.IncrementRequests()
+	}
+	recordFailed := func() {
+		recordedCurrent = true
+		countAccounted(1)
+		request.options.Progress.IncrementFailedRequestsBy(1)
+	}
+	recordStopped := func() {
+		remaining := int64(generator.Remaining())
+		if !recordedCurrent {
+			remaining++
+		}
+		recordSkipped(remaining)
+	}
 
+	initialPrevious := cloneInternalEvent(previous)
 	initialContext := dynamicValueContext{
 		values:   generators.CopyMap(dynamicValues),
-		previous: cloneInternalEvent(previous),
+		previous: cloneInternalEvent(initialPrevious),
 	}
 	dynamicValueContexts := []dynamicValueContext{initialContext}
 	if request.IterateAll {
 		dynamicValueContexts = expandDynamicValueContexts(initialContext, generators.MergeMapsMany(dynamicValues), true)
 	}
+	oncePrevious := make(output.InternalEvent)
+	lastRequestIndex := 0
 	var requestErr error
 
 executionLoop:
 	for {
 		executeFunc := func(data string, payloads, dynamicValue map[string]interface{}, previousEvent output.InternalEvent) (skipBranch, stop bool, extracted map[string][]string, err error) {
+			recordedCurrent = false
 			hasInteractMatchers := interactsh.HasMatchers(request.CompiledOperators)
 
 			ctx := request.newContext(input)
-			ctxWithTimeout, cancel := context.WithTimeoutCause(ctx, request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
-			defer cancel()
-
-			generatedHttpRequest, err := generator.Make(ctxWithTimeout, input, data, payloads, dynamicValue)
+			generatedHttpRequest, err := generator.Make(ctx, input, data, payloads, dynamicValue)
 			if err != nil {
 				if err == types.ErrNoMoreRequests {
 					return false, true, nil, nil
@@ -679,14 +730,22 @@ executionLoop:
 			// a copy and using it to check for host errors etc
 			// but this should be replaced once templateCtx is refactored properly
 			updatedInput := contextargs.GetCopyIfHostOutdated(input, generatedHttpRequest.URL())
+			if generatedHttpRequest.customCancelFunction != nil {
+				defer generatedHttpRequest.customCancelFunction()
+			}
+			if request.isUnresponsiveAddress(updatedInput) {
+				return false, true, nil, nil
+			}
 
 			// Full URL, so https and an explicit port stay on the same key the observation uses.
 			if err := request.rateLimitTake(ctx, hostBackoffKey(generatedHttpRequest, input)); err != nil {
 				return false, true, nil, err
 			}
-
-			if generatedHttpRequest.customCancelFunction != nil {
-				defer generatedHttpRequest.customCancelFunction()
+			if generatedHttpRequest.request != nil && generatedHttpRequest.customCancelFunction == nil {
+				// Rate admission must not consume the network request's timeout.
+				timedCtx, cancel := context.WithTimeoutCause(generatedHttpRequest.request.Context(), request.options.Options.GetTimeouts().HttpTimeout, ErrHttpEngineRequestDeadline)
+				defer cancel()
+				generatedHttpRequest.request.WithContext(timedCtx)
 			}
 
 			hasInteractMarkers := interactsh.HasMarkers(data) || len(generatedHttpRequest.interactshURLs) > 0
@@ -737,6 +796,9 @@ executionLoop:
 			if errors.Is(execReqErr, ErrMissingVars) {
 				return true, false, nil, nil
 			}
+			if stopOnMatch != nil && stopOnMatch.Load() && errors.Is(execReqErr, context.Canceled) {
+				return false, true, nil, nil
+			}
 
 			request.recordHostResult(updatedInput, execReqErr)
 			if execReqErr != nil {
@@ -744,9 +806,9 @@ executionLoop:
 				reqKitErr.Msgf("got err while executing %v", generatedHttpRequest.URL())
 
 				requestErr = reqKitErr
-				request.options.Progress.IncrementFailedRequestsBy(1)
+				recordFailed()
 			} else {
-				request.options.Progress.IncrementRequests()
+				recordSent()
 			}
 
 			// If this was a match, and we want to stop at first match, skip all further requests.
@@ -761,6 +823,13 @@ executionLoop:
 		if !ok {
 			break
 		}
+		if generator.payloadIterator != nil && generator.currentIndex <= lastRequestIndex {
+			for i := range dynamicValueContexts {
+				dynamicValueContexts[i].previous = cloneInternalEvent(initialPrevious)
+				maps.Copy(dynamicValueContexts[i].previous, oncePrevious)
+			}
+		}
+		lastRequestIndex = generator.currentIndex
 
 		select {
 		case <-input.Context().Done():
@@ -788,11 +857,11 @@ executionLoop:
 					if requestErr == nil {
 						requestErr = gotErr
 					}
-					request.options.Progress.SetRequests(uint64(generator.Remaining() + 1))
+					recordSkipped(int64(generator.Remaining() + 1))
 					break executionLoop
 				}
 				if stop {
-					request.options.Progress.SetRequests(uint64(generator.Remaining() + 1))
+					recordStopped()
 					break executionLoop
 				}
 				if skipBranch {
@@ -807,6 +876,7 @@ executionLoop:
 				dynamicValueContexts = nil
 				continue
 			}
+			maps.Copy(oncePrevious, oncePreviousDelta)
 
 			nextDynamicValueContexts := make([]dynamicValueContext, 0, len(dynamicValueContexts))
 			for _, branch := range dynamicValueContexts {
@@ -830,11 +900,11 @@ executionLoop:
 				if requestErr == nil {
 					requestErr = gotErr
 				}
-				request.options.Progress.SetRequests(uint64(generator.Remaining() + 1))
+				recordSkipped(int64(generator.Remaining() + 1))
 				break executionLoop
 			}
 			if stop {
-				request.options.Progress.SetRequests(uint64(generator.Remaining() + 1))
+				recordStopped()
 				break executionLoop
 			}
 			if skipBranch {
@@ -847,6 +917,119 @@ executionLoop:
 	return requestErr
 }
 
+func (request *Request) canRunDependentPayloadsConcurrently() bool {
+	if request.IterateAll || request.Race || request.Pipeline || len(request.Fuzzing) > 0 || len(request.Raw) < 2 || interactsh.HasMatchers(request.CompiledOperators) {
+		return false
+	}
+	for _, raw := range request.Raw {
+		if interactsh.HasMarkers(raw) {
+			return false
+		}
+		if mark, hasMark := parseFlowAnnotations(raw); hasMark && mark == Once {
+			return false
+		}
+	}
+	return true
+}
+
+func (request *Request) executeDependentPayloads(input *contextargs.Context, dynamicValues, previous output.InternalEvent, callback protocols.OutputEventCallback) error {
+	workers := min(request.dependentPayloadThreads, defaultMaxWorkers)
+	if request.options.Options.PayloadConcurrency > 0 {
+		workers = min(workers, request.options.Options.PayloadConcurrency)
+	}
+	workerCtx, cancel := context.WithCancel(input.Context())
+	defer cancel()
+	permits := semaphore.NewWeighted(int64(workers))
+
+	shouldStop := request.options.Options.StopAtFirstMatch || request.StopAtFirstMatch || request.options.StopAtFirstMatch
+	var found atomic.Bool
+	var callbackMu sync.Mutex
+	wrappedCallback := func(event *output.InternalWrappedEvent) {
+		callbackMu.Lock()
+		defer callbackMu.Unlock()
+		if shouldStop && found.Load() {
+			return
+		}
+		callback(event)
+		if shouldStop && event.OperatorsResult != nil && event.OperatorsResult.Matched {
+			found.Store(true)
+			cancel()
+		}
+	}
+
+	tasks := make(chan map[string]interface{}, workers)
+	var workersWG sync.WaitGroup
+	var errorsMu sync.Mutex
+	var executionErrors []error
+	var accounted atomic.Int64
+	for i := 0; i < workers; i++ {
+		workersWG.Add(1)
+		go func() {
+			defer workersWG.Done()
+			for payloads := range tasks {
+				if workerCtx.Err() != nil {
+					return
+				}
+				weight := int64(1)
+				if protocolstate.IsLowOnMemory() {
+					allowed := max(1, protocolstate.GuardThreadsOrDefault(workers))
+					weight = int64((workers + allowed - 1) / allowed)
+				}
+				if err := permits.Acquire(workerCtx, weight); err != nil {
+					return
+				}
+				generator := &requestGenerator{
+					request:         request,
+					options:         request.options,
+					currentPayloads: payloads,
+					onceFlow:        make(map[string]struct{}),
+				}
+				groupInput := input.CloneWithContext(workerCtx)
+				err := request.executeSequentialHTTP(groupInput, dynamicValues, previous, wrappedCallback, generator, &accounted, &found)
+				permits.Release(weight)
+				if err != nil {
+					if found.Load() && errors.Is(err, context.Canceled) {
+						continue
+					}
+					errorsMu.Lock()
+					executionErrors = append(executionErrors, err)
+					errorsMu.Unlock()
+				}
+			}
+		}()
+	}
+
+	iterator := request.generator.NewIterator()
+enqueue:
+	for {
+		payloads, ok := iterator.Value()
+		if !ok {
+			break
+		}
+		select {
+		case <-workerCtx.Done():
+			break enqueue
+		case tasks <- payloads:
+		}
+	}
+	close(tasks)
+	workersWG.Wait()
+	missing := int64(request.Requests()) - accounted.Load()
+	if missing < 0 {
+		return fmt.Errorf("dependent payload request accounting exceeded planned requests by %d", -missing)
+	}
+	if missing > 0 {
+		request.options.Progress.SetRequests(uint64(missing))
+	}
+	if found.Load() {
+		return nil
+	}
+	if err := input.Context().Err(); err != nil {
+		return err
+	}
+	return multierr.Combine(executionErrors...)
+}
+
 type dynamicValueContext struct {
 	values   map[string]interface{}
 	seen     map[string]map[string]struct{}
@@ -1247,6 +1430,13 @@ func (request *Request) executeRequest(input *contextargs.Context, generatedRequ
 	// Record before dump or the dialer lookup can return and drop the outcome.
 	if sent {
 		request.observeHostBackoff(backoffKey, resp, err)
+		if observer, ok := request.options.Progress.(httpRequestTimingObserver); ok {
+			statusCode := 0
+			if resp != nil {
+				statusCode = resp.StatusCode
+			}
+			observer.ObserveHTTPRequest(time.Since(timeStart), statusCode, err)
+		}
 	}
 	// use request url as matched url if empty
 	if formedURL == "" {

pkg/protocols/http/request_condition_test.go

@@ -2,7 +2,15 @@ package http
 
 import (
 	"context"
+	"crypto/md5"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -13,9 +21,595 @@ import (
 	extractorpkg "github.com/projectdiscovery/nuclei/v3/pkg/operators/extractors"
 	matcherpkg "github.com/projectdiscovery/nuclei/v3/pkg/operators/matchers"
 	"github.com/projectdiscovery/nuclei/v3/pkg/output"
+	"github.com/projectdiscovery/nuclei/v3/pkg/progress"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/generators"
+	"github.com/projectdiscovery/nuclei/v3/pkg/types"
+	"github.com/projectdiscovery/utils/memguardian"
 )
 
+func TestIndexedResponsePayloadCannotUseIndependentRequestThreads(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requests.Add(1)
+		if request.URL.Path == "/first" {
+			writer.WriteHeader(http.StatusCreated)
+			return
+		}
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID: "indexed-response-payload-threading",
+		Raw: []string{
+			"GET /first?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+		},
+		Threads:  10,
+		Payloads: map[string]interface{}{"probe": []string{"one", "two"}},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{"status_code_1 == 201 && status_code_2 == 200"},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, request.Compile(executerOptions))
+	require.Zero(t, request.Threads)
+
+	run := func() int {
+		matched := 0
+		err := request.ExecuteWithResults(
+			contextargs.NewWithInput(t.Context(), server.URL),
+			make(output.InternalEvent),
+			make(output.InternalEvent),
+			func(event *output.InternalWrappedEvent) {
+				if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+					matched++
+				}
+			},
+		)
+		require.NoError(t, err)
+		return matched
+	}
+
+	require.Equal(t, 2, run())
+	require.Equal(t, int32(4), requests.Load())
+
+	request.Threads = 10
+	requests.Store(0)
+	require.Zero(t, run())
+	require.Equal(t, int32(4), requests.Load())
+}
+
+func TestDependentPayloadThreadsRejectSharedRequestState(t *testing.T) {
+	newRequest := func() *Request {
+		return &Request{Raw: []string{
+			"GET /first HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second HTTP/1.1\nHost: {{Hostname}}",
+		}}
+	}
+
+	base := newRequest()
+	require.True(t, base.canRunDependentPayloadsConcurrently())
+
+	iterating := newRequest()
+	iterating.IterateAll = true
+	require.False(t, iterating.canRunDependentPayloadsConcurrently())
+
+	racing := newRequest()
+	racing.Race = true
+	require.False(t, racing.canRunDependentPayloadsConcurrently())
+
+	pipelined := newRequest()
+	pipelined.Pipeline = true
+	require.False(t, pipelined.canRunDependentPayloadsConcurrently())
+
+	once := newRequest()
+	once.Raw[0] = "@once\n" + once.Raw[0]
+	require.False(t, once.canRunDependentPayloadsConcurrently())
+}
+
+func TestIndexedResponsePayloadThreadsKeepPairsOrdered(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(false)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var firstStarted atomic.Int32
+	var firstActive atomic.Int32
+	var maximumFirstActive atomic.Int32
+	var outOfOrder atomic.Int32
+	firstGate := make(chan struct{})
+	firstDone := make(map[string]bool)
+	var firstDoneMu sync.Mutex
+	negativeProbe := fmt.Sprintf("%x", md5.Sum([]byte("two")))
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		probe := request.URL.Query().Get("probe")
+		switch request.URL.Path {
+		case "/first":
+			active := firstActive.Add(1)
+			for old := maximumFirstActive.Load(); active > old && !maximumFirstActive.CompareAndSwap(old, active); old = maximumFirstActive.Load() {
+			}
+			if firstStarted.Add(1) == 2 {
+				close(firstGate)
+			}
+			select {
+			case <-firstGate:
+			case <-time.After(250 * time.Millisecond):
+			}
+			firstDoneMu.Lock()
+			firstDone[probe] = true
+			firstDoneMu.Unlock()
+			firstActive.Add(-1)
+			writer.WriteHeader(http.StatusCreated)
+		case "/second":
+			firstDoneMu.Lock()
+			if !firstDone[probe] {
+				outOfOrder.Add(1)
+			}
+			firstDoneMu.Unlock()
+			if probe == negativeProbe {
+				writer.WriteHeader(http.StatusNotFound)
+				return
+			}
+			writer.WriteHeader(http.StatusOK)
+		}
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID: "indexed-response-payload-pairs",
+		Raw: []string{
+			"GET /first?probe={{md5(probe)}} HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second?probe={{md5(probe)}} HTTP/1.1\nHost: {{Hostname}}",
+		},
+		PayloadThreads: 2,
+		Payloads:       map[string]interface{}{"probe": []string{"one", "two"}},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{"status_code_1 == 201 && status_code_2 == 200"},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, request.Compile(executerOptions))
+
+	var matches atomic.Int32
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matches.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(1), matches.Load())
+	require.Zero(t, outOfOrder.Load())
+	require.Equal(t, int32(2), maximumFirstActive.Load())
+	require.Len(t, firstDone, 2)
+}
+
+func TestDependentPayloadRateLimitWaitDoesNotExpireHTTPTimeout(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	options.PerHostRateLimit = true
+	options.RateLimit = 1
+	options.RateLimitDuration = 80 * time.Millisecond
+	options.SetTimeouts(&types.Timeouts{HttpTimeout: 50 * time.Millisecond})
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	var sent atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		sent.Add(1)
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID: "dependent-payload-rate-wait",
+		Raw: []string{
+			"GET /first?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+		},
+		PayloadThreads: 2,
+		Payloads: map[string]interface{}{
+			"probe": []string{"one", "two"},
+		},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{"status_code_1 == 200 && status_code_2 == 200"},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, request.Compile(executerOptions))
+
+	var matches atomic.Int32
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matches.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(4), sent.Load())
+	require.Equal(t, int32(2), matches.Load())
+}
+
+func TestDependentPayloadRateLimitedLastMatchAccounting(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	options.PerHostRateLimit = true
+	options.RateLimit = 1
+	options.RateLimitDuration = 20 * time.Millisecond
+	options.SetTimeouts(&types.Timeouts{HttpTimeout: 15 * time.Millisecond})
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	var sent atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		sent.Add(1)
+		if request.URL.Path == "/first" {
+			writer.WriteHeader(http.StatusCreated)
+			return
+		}
+		if request.URL.Query().Get("probe") == "key-50" {
+			writer.WriteHeader(http.StatusOK)
+			return
+		}
+		writer.WriteHeader(http.StatusNotFound)
+	}))
+	defer server.Close()
+
+	keys := make([]string, 51)
+	for i := range keys {
+		keys[i] = fmt.Sprintf("key-%d", i)
+	}
+	request := &Request{
+		ID: "dependent-payload-last-match",
+		Raw: []string{
+			"GET /first?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+		},
+		PayloadThreads:   10,
+		StopAtFirstMatch: true,
+		Payloads:         map[string]interface{}{"probe": keys},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{"status_code_1 == 201 && status_code_2 == 200"},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	progressCounter := &payloadRequestProgress{Progress: executerOptions.Progress}
+	executerOptions.Progress = progressCounter
+	require.NoError(t, request.Compile(executerOptions))
+
+	var matches atomic.Int32
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matches.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(1), matches.Load())
+	require.GreaterOrEqual(t, sent.Load(), int32(2))
+	require.LessOrEqual(t, sent.Load(), int32(102))
+	require.Equal(t, int64(sent.Load()), progressCounter.sent.Load())
+	require.Equal(t, int64(102), progressCounter.sent.Load()+progressCounter.skipped.Load()+progressCounter.failed.Load())
+	require.Zero(t, progressCounter.failed.Load())
+}
+
+func TestIndexedResponsePayloadThreadsStopAfterFirstMatch(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(false)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requests.Add(1)
+		if request.URL.Path == "/first" {
+			writer.WriteHeader(http.StatusCreated)
+			return
+		}
+		if request.URL.Query().Get("probe") == "match" {
+			writer.WriteHeader(http.StatusOK)
+			return
+		}
+		writer.WriteHeader(http.StatusNotFound)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID: "indexed-response-stop-first-match",
+		Raw: []string{
+			"GET /first?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+			"GET /second?probe={{probe}} HTTP/1.1\nHost: {{Hostname}}",
+		},
+		PayloadThreads:   2,
+		StopAtFirstMatch: true,
+		Payloads: map[string]interface{}{
+			"probe": []string{"match", "miss-a", "miss-b", "miss-c"},
+		},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{"status_code_1 == 201 && status_code_2 == 200"},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	progressCounter := &payloadRequestProgress{Progress: executerOptions.Progress}
+	executerOptions.Progress = progressCounter
+	require.NoError(t, request.Compile(executerOptions))
+
+	var matches atomic.Int32
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matches.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(1), matches.Load())
+	require.Less(t, requests.Load(), int32(8))
+	require.Equal(t, int64(request.Requests()), progressCounter.sent.Load()+progressCounter.skipped.Load()+progressCounter.failed.Load())
+	require.Zero(t, progressCounter.failed.Load())
+}
+
+func TestIndexedResponsePayloadThreadsCoverAllFiveHundredResponses(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(false)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var active atomic.Int32
+	var maximum atomic.Int32
+	var served atomic.Int32
+	var keysMu sync.Mutex
+	keys := make(map[string]int)
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		current := active.Add(1)
+		for old := maximum.Load(); current > old && !maximum.CompareAndSwap(old, current); old = maximum.Load() {
+		}
+		key := strings.SplitN(request.URL.RawQuery, "=", 2)[0]
+		keysMu.Lock()
+		keys[key]++
+		keysMu.Unlock()
+		time.Sleep(5 * time.Millisecond)
+		active.Add(-1)
+		served.Add(1)
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	buckets := make([]string, 17)
+	for i := range buckets {
+		buckets[i] = fmt.Sprintf("bucket-%d", i)
+	}
+	for _, tc := range []struct {
+		name           string
+		payloadThreads int
+		minimumActive  int32
+		maximumActive  int32
+		lowMemory      bool
+	}{
+		{name: "serial", minimumActive: 1, maximumActive: 1},
+		{name: "guarded", payloadThreads: 10, minimumActive: 1, maximumActive: 2, lowMemory: true},
+		{name: "grouped", payloadThreads: 10, minimumActive: 5},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			memguardian.DefaultMemGuardian.Warning.Store(tc.lowMemory)
+			maximum.Store(0)
+			served.Store(0)
+			keysMu.Lock()
+			clear(keys)
+			keysMu.Unlock()
+			request := &Request{
+				ID: "indexed-response-5xx-coverage-" + tc.name,
+				Raw: []string{
+					"GET /first?{{rand}}-{{md5(escape)}}-{{bucket}}=x HTTP/1.1\nHost: {{Hostname}}",
+					"GET /second?{{rand}}-{{md5(escape)}}-{{bucket}}=x HTTP/1.1\nHost: {{Hostname}}",
+				},
+				PayloadThreads: tc.payloadThreads,
+				AttackType:     generators.AttackTypeHolder{Value: generators.ClusterBombAttack},
+				Payloads: map[string]interface{}{
+					"escape": []string{"\v", "\f", "\x1c", "\x1d", "\x1e", "\x1f"},
+					"bucket": buckets,
+				},
+				Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+					Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+					DSL:  []string{"status_code_2 == 200"},
+				}),
+				StopAtFirstMatch: true,
+			}
+			executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+				ID:   request.ID,
+				Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+			})
+			executerOptions.RateLimiter = nil
+			require.NoError(t, executerOptions.Variables.UnmarshalJSON([]byte(`{"rand":"{{rand_base(5)}}"}`)))
+			progressCounter := &payloadRequestProgress{Progress: executerOptions.Progress}
+			executerOptions.Progress = progressCounter
+			require.NoError(t, request.Compile(executerOptions))
+
+			var matches atomic.Int32
+			require.NoError(t, request.ExecuteWithResults(
+				contextargs.NewWithInput(t.Context(), server.URL),
+				make(output.InternalEvent),
+				make(output.InternalEvent),
+				func(event *output.InternalWrappedEvent) {
+					if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+						matches.Add(1)
+					}
+				},
+			))
+			require.Zero(t, matches.Load())
+			require.Equal(t, int32(204), served.Load())
+			require.Equal(t, int64(204), progressCounter.sent.Load())
+			require.Zero(t, progressCounter.skipped.Load())
+			require.Zero(t, progressCounter.failed.Load())
+			require.GreaterOrEqual(t, maximum.Load(), tc.minimumActive)
+			if tc.maximumActive > 0 {
+				require.LessOrEqual(t, maximum.Load(), tc.maximumActive)
+			}
+			keysMu.Lock()
+			require.Len(t, keys, 102)
+			for _, count := range keys {
+				require.Equal(t, 2, count)
+			}
+			keysMu.Unlock()
+		})
+	}
+}
+
+func TestIndexedResponsePayloadThreadsPreserveCachePoisoningDetection(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(false)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var mu sync.Mutex
+	cache := make(map[string]string)
+	var served atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		served.Add(1)
+		key := strings.SplitN(request.URL.RawQuery, "=", 2)[0]
+		mu.Lock()
+		defer mu.Unlock()
+		if request.URL.Path == "/first" {
+			if request.Header.Get("X-Bucket") == "positive" {
+				cache[key] = "alert(document.domain)"
+			}
+			writer.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		if value, ok := cache[key]; ok {
+			writer.WriteHeader(http.StatusOK)
+			_, _ = writer.Write([]byte(value))
+			return
+		}
+		writer.WriteHeader(http.StatusNotFound)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID: "indexed-response-cache-poisoning",
+		Raw: []string{
+			"GET /first?{{rand}}-{{md5(escape)}}-{{bucket}}=x HTTP/1.1\nHost: {{Hostname}}\nX-Bucket: {{bucket}}",
+			"GET /second?{{rand}}-{{md5(escape)}}-{{bucket}}=x HTTP/1.1\nHost: {{Hostname}}",
+		},
+		PayloadThreads: 2,
+		AttackType:     generators.AttackTypeHolder{Value: generators.ClusterBombAttack},
+		Payloads: map[string]interface{}{
+			"escape": []string{"\v"},
+			"bucket": []string{"positive", "negative"},
+		},
+		Operators: operatorsWithMatchers(&matcherpkg.Matcher{
+			Type: matcherpkg.MatcherTypeHolder{MatcherType: matcherpkg.DSLMatcher},
+			DSL:  []string{`contains(body_2, "alert(document.domain)") && status_code_2 == 200`},
+		}),
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, executerOptions.Variables.UnmarshalJSON([]byte(`{"rand":"{{rand_base(5)}}"}`)))
+	require.NoError(t, request.Compile(executerOptions))
+
+	var matches atomic.Int32
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matches.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(4), served.Load())
+	require.Equal(t, int32(1), matches.Load())
+	require.Len(t, cache, 1)
+}
+
+type payloadRequestProgress struct {
+	progress.Progress
+	sent    atomic.Int64
+	skipped atomic.Int64
+	failed  atomic.Int64
+}
+
+func (p *payloadRequestProgress) IncrementRequests() {
+	p.sent.Add(1)
+	p.Progress.IncrementRequests()
+}
+
+func (p *payloadRequestProgress) SetRequests(count uint64) {
+	p.skipped.Add(int64(count))
+	p.Progress.SetRequests(count)
+}
+
+func (p *payloadRequestProgress) IncrementFailedRequestsBy(count int64) {
+	p.failed.Add(count)
+	p.Progress.IncrementFailedRequestsBy(count)
+}
+
 func TestNeedsRequestConditionDurationReferences(t *testing.T) {
 	tests := []struct {
 		name    string

pkg/protocols/http/request_test.go

@@ -21,10 +21,15 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/operators/extractors"
 	"github.com/projectdiscovery/nuclei/v3/pkg/operators/matchers"
 	"github.com/projectdiscovery/nuclei/v3/pkg/output"
+	"github.com/projectdiscovery/nuclei/v3/pkg/progress"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/hostbackoff"
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/interactsh"
+	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/protocolstate"
 	"github.com/projectdiscovery/ratelimit"
 	"github.com/projectdiscovery/retryablehttp-go"
+	"github.com/projectdiscovery/utils/memguardian"
 	"io"
 )
 
@@ -1296,6 +1301,291 @@ func TestExecuteParallelHTTP_StopAtFirstMatchCancelsQueuedRateLimitWaiters(t *te
 	require.Equal(t, int32(1), requests.Load())
 }
 
+func TestPayloadConcurrencyIsGuardedOnceUnderMemoryPressure(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(true)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var active atomic.Int32
+	var maximum atomic.Int32
+	var total atomic.Int32
+	var matched atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		current := active.Add(1)
+		for previous := maximum.Load(); current > previous && !maximum.CompareAndSwap(previous, current); previous = maximum.Load() {
+		}
+		time.Sleep(40 * time.Millisecond)
+		active.Add(-1)
+		total.Add(1)
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID:      "memory-guard-payload-concurrency",
+		Method:  HTTPMethodTypeHolder{MethodType: HTTPGet},
+		Path:    []string{"{{BaseURL}}/test?payload={{v}}"},
+		Threads: 20,
+		Payloads: map[string]interface{}{
+			"v": []string{"1", "2", "3", "4", "5", "6", "7", "8"},
+		},
+		Operators: operators.Operators{Matchers: []*matchers.Matcher{{
+			Type: matchers.MatcherTypeHolder{MatcherType: matchers.DSLMatcher},
+			DSL:  []string{"status_code == 200"},
+		}}},
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, request.Compile(executerOptions))
+	compiledThreads := request.Threads
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matched.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(8), total.Load())
+	require.Equal(t, int32(8), matched.Load())
+	if compiledThreads != 20 || maximum.Load() <= 1 {
+		t.Fatalf("memory guard applied more than once: compiled threads=%d, maximum concurrent requests=%d", compiledThreads, maximum.Load())
+	}
+
+	memguardian.DefaultMemGuardian.Warning.Store(false)
+	maximum.Store(0)
+	total.Store(0)
+	matched.Store(0)
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(event *output.InternalWrappedEvent) {
+			if event.OperatorsResult != nil && event.OperatorsResult.Matched {
+				matched.Add(1)
+			}
+		},
+	))
+	require.Equal(t, int32(8), total.Load())
+	require.Equal(t, int32(8), matched.Load())
+	require.Greater(t, maximum.Load(), int32(4))
+}
+
+func TestDefaultPayloadConcurrencyRespectsMemoryPressure(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	options.PayloadConcurrency = 25
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+	require.NotNil(t, memguardian.DefaultMemGuardian)
+	previousWarning := memguardian.DefaultMemGuardian.Warning.Swap(true)
+	t.Cleanup(func() { memguardian.DefaultMemGuardian.Warning.Store(previousWarning) })
+
+	var active atomic.Int32
+	var maximum atomic.Int32
+	var total atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		current := active.Add(1)
+		for previous := maximum.Load(); current > previous && !maximum.CompareAndSwap(previous, current); previous = maximum.Load() {
+		}
+		time.Sleep(40 * time.Millisecond)
+		active.Add(-1)
+		total.Add(1)
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID:      "default-payload-concurrency-under-memory-pressure",
+		Method:  HTTPMethodTypeHolder{MethodType: HTTPGet},
+		Path:    []string{"{{BaseURL}}/test?payload={{v}}"},
+		Threads: options.PayloadConcurrency,
+		Payloads: map[string]interface{}{
+			"v": []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16"},
+		},
+		Operators: operators.Operators{Matchers: []*matchers.Matcher{{
+			Type: matchers.MatcherTypeHolder{MatcherType: matchers.DSLMatcher},
+			DSL:  []string{"status_code == 200"},
+		}}},
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.RateLimiter = nil
+	require.NoError(t, request.Compile(executerOptions))
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(*output.InternalWrappedEvent) {},
+	))
+	require.Equal(t, int32(16), total.Load())
+	require.LessOrEqual(t, maximum.Load(), int32(protocolstate.GuardThreadsOrDefault(options.PayloadConcurrency)))
+	require.Greater(t, maximum.Load(), int32(1))
+}
+
+type timingProgressObserver struct {
+	progress.Progress
+	mu              sync.Mutex
+	requestDuration time.Duration
+	requestStatus   int
+	requestError    error
+	requestCount    int
+	rateWaitCount   int
+	rateWait        time.Duration
+}
+
+func (o *timingProgressObserver) ObserveHTTPRateLimitWait(duration time.Duration) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.rateWaitCount++
+	o.rateWait = duration
+}
+
+func (o *timingProgressObserver) ObserveHTTPRequest(duration time.Duration, status int, err error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.requestDuration = duration
+	o.requestStatus = status
+	o.requestError = err
+	o.requestCount++
+}
+
+func TestHTTPExecutionObservesRequestTiming(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		time.Sleep(20 * time.Millisecond)
+		writer.WriteHeader(http.StatusTeapot)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID:     "http-timing-observer",
+		Method: HTTPMethodTypeHolder{MethodType: HTTPGet},
+		Path:   []string{"{{BaseURL}}/"},
+		Operators: operators.Operators{Matchers: []*matchers.Matcher{{
+			Type:   matchers.MatcherTypeHolder{MatcherType: matchers.StatusMatcher},
+			Status: []int{http.StatusTeapot},
+		}}},
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	observer := &timingProgressObserver{Progress: executerOptions.Progress}
+	executerOptions.Progress = observer
+	require.NoError(t, request.Compile(executerOptions))
+	require.NoError(t, request.ExecuteWithResults(
+		contextargs.NewWithInput(t.Context(), server.URL),
+		make(output.InternalEvent),
+		make(output.InternalEvent),
+		func(*output.InternalWrappedEvent) {},
+	))
+
+	observer.mu.Lock()
+	defer observer.mu.Unlock()
+	require.Equal(t, 1, observer.rateWaitCount)
+	require.Equal(t, 1, observer.requestCount)
+	require.Equal(t, http.StatusTeapot, observer.requestStatus)
+	require.NoError(t, observer.requestError)
+	require.GreaterOrEqual(t, observer.requestDuration, 20*time.Millisecond)
+}
+
+func TestHTTPRateLimitWaitExcludesHostBackoff(t *testing.T) {
+	backoff := hostbackoff.New(hostbackoff.Config{Step: 200 * time.Millisecond, Max: 200 * time.Millisecond})
+	observer := &timingProgressObserver{}
+	request := &Request{options: &protocols.ExecutorOptions{HostBackoff: backoff, Progress: observer}}
+	target := "https://example.com/path"
+	request.observeHostBackoff(target, &http.Response{StatusCode: http.StatusTooManyRequests, Header: http.Header{}}, nil)
+
+	started := time.Now()
+	require.NoError(t, request.rateLimitTake(t.Context(), target))
+	elapsed := time.Since(started)
+
+	observer.mu.Lock()
+	defer observer.mu.Unlock()
+	require.Equal(t, 1, observer.rateWaitCount)
+	require.GreaterOrEqual(t, elapsed, 150*time.Millisecond)
+	require.Less(t, observer.rateWait, elapsed/2)
+}
+
+func TestHTTPRateLimitWaitNotObservedWhenBackoffCanceled(t *testing.T) {
+	backoff := hostbackoff.New(hostbackoff.Config{Step: time.Second, Max: time.Second})
+	observer := &timingProgressObserver{}
+	request := &Request{options: &protocols.ExecutorOptions{HostBackoff: backoff, Progress: observer}}
+	target := "https://example.com/path"
+	request.observeHostBackoff(target, &http.Response{StatusCode: http.StatusTooManyRequests, Header: http.Header{}}, nil)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	require.ErrorIs(t, request.rateLimitTake(ctx, target), context.Canceled)
+	observer.mu.Lock()
+	defer observer.mu.Unlock()
+	require.Zero(t, observer.rateWaitCount)
+}
+
+func TestSerialHTTPUnresponsiveHostSkipsBeforeRateLimit(t *testing.T) {
+	options := testutils.DefaultOptions.Copy()
+	options.ExecutionId = t.Name()
+	options.RateLimit = 1
+	options.RateLimitDuration = 2 * time.Second
+	options.RestrictLocalNetworkAccess = false
+	testutils.Init(options)
+	t.Cleanup(func() { testutils.Cleanup(options) })
+
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		requests.Add(1)
+		writer.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	request := &Request{
+		ID:     "serial-skip-before-rate-limit",
+		Method: HTTPMethodTypeHolder{MethodType: HTTPGet},
+		Path:   []string{"{{BaseURL}}/"},
+	}
+	executerOptions := testutils.NewMockExecuterOptions(options, &testutils.TemplateInfo{
+		ID:   request.ID,
+		Info: model.Info{SeverityHolder: severity.Holder{Severity: severity.Low}, Name: "test"},
+	})
+	executerOptions.HostErrorsCache = &fakeHostErrorsCache{}
+	observer := &timingProgressObserver{Progress: executerOptions.Progress}
+	executerOptions.Progress = observer
+	executerOptions.RateLimiter = ratelimit.New(context.Background(), 1, 2*time.Second)
+	t.Cleanup(executerOptions.RateLimiter.Stop)
+	require.NoError(t, executerOptions.RateLimiter.TakeContext(t.Context()))
+	require.NoError(t, request.Compile(executerOptions))
+
+	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+	defer cancel()
+	started := time.Now()
+	err := request.ExecuteWithResults(contextargs.NewWithInput(ctx, server.URL), make(output.InternalEvent), make(output.InternalEvent), func(*output.InternalWrappedEvent) {})
+	require.NoError(t, err)
+	require.Less(t, time.Since(started), 250*time.Millisecond)
+	require.Zero(t, requests.Load())
+	observer.mu.Lock()
+	defer observer.mu.Unlock()
+	require.Zero(t, observer.rateWaitCount)
+}
+
 func TestExecuteParallelHTTP_SkipOnUnresponsiveFromCache(t *testing.T) {
 	options := testutils.DefaultOptions
 	testutils.Init(options)

pkg/tmplexec/flow/flow_executor_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/julienschmidt/httprouter"
+	"github.com/projectdiscovery/nuclei/v3/internal/tests/testutils"
 	"github.com/projectdiscovery/nuclei/v3/pkg/catalog/config"
 	"github.com/projectdiscovery/nuclei/v3/pkg/catalog/disk"
 	"github.com/projectdiscovery/nuclei/v3/pkg/loader/workflow"
@@ -21,7 +22,6 @@ import (
 	"github.com/projectdiscovery/nuclei/v3/pkg/protocols/common/contextargs"
 	"github.com/projectdiscovery/nuclei/v3/pkg/scan"
 	"github.com/projectdiscovery/nuclei/v3/pkg/templates"
-	"github.com/projectdiscovery/nuclei/v3/internal/tests/testutils"
 	"github.com/projectdiscovery/ratelimit"
 	"github.com/stretchr/testify/require"
 )
@@ -287,6 +287,10 @@ func newMultiPayloadServer() *httptest.Server {
 // survive across multiple pitchfork payload iterations (generator restarts).
 func TestFlowMultiPayloadIteration(t *testing.T) {
 	setup()
+	options := executerOpts.Options
+	previousConcurrency := options.PayloadConcurrency
+	options.PayloadConcurrency = 2
+	t.Cleanup(func() { options.PayloadConcurrency = previousConcurrency })
 	ts := newMultiPayloadServer()
 	defer ts.Close()
 
@@ -295,6 +299,7 @@ func TestFlowMultiPayloadIteration(t *testing.T) {
 
 	err = tmpl.Executer.Compile()
 	require.Nil(t, err, "could not compile template")
+	require.Zero(t, tmpl.RequestsHTTP[0].Threads, "request-condition payloads require ordered execution")
 
 	input := contextargs.NewWithInput(context.Background(), ts.URL)
 	ctx := scan.NewScanContext(context.Background(), input)
@@ -303,6 +308,24 @@ func TestFlowMultiPayloadIteration(t *testing.T) {
 	require.True(t, gotresults, "expected match with flow + multi payload iteration")
 }
 
+func TestIndependentMultiPayloadRequestsRemainParallel(t *testing.T) {
+	setup()
+	options := executerOpts.Options
+	previousConcurrency := options.PayloadConcurrency
+	options.PayloadConcurrency = 2
+	t.Cleanup(func() { options.PayloadConcurrency = previousConcurrency })
+
+	tmpl, err := templates.Parse("testcases/flow-multi-payload-iteration.yaml", nil, executerOpts)
+	require.NoError(t, err)
+	for _, matcher := range tmpl.RequestsHTTP[0].Matchers {
+		matcher.Part = "body"
+	}
+	require.False(t, tmpl.RequestsHTTP[0].NeedsRequestCondition())
+
+	require.NoError(t, tmpl.Executer.Compile())
+	require.Positive(t, tmpl.RequestsHTTP[0].Threads)
+}
+
 // TestFlowRequestConditionWithPayloads is the same as above but with
 // pitchfork payloads and header-based extraction, matching the exact
 // pattern from issue #5095.