tests/php_test_files/jobs/jobs_ok_headers.php (3)

1-15: LGTM! Consider adding strict types declaration.

The file setup and imports are well-organized. For better type safety, consider adding strict types declaration.

 <?php
+declare(strict_types=1);

---

29-31: Enhance error reporting for debugging.

The current error handling only sends the exception message. Consider including more context for debugging.

     } catch (\Throwable $e) {
-        $task->error((string)$e);
+        $task->error(sprintf(
+            "Error processing task [%s]: %s\nTrace: %s",
+            $task->getId(),
+            $e->getMessage(),
+            $e->getTraceAsString()
+        ));
     }

---

18-32: Add graceful shutdown handling.

The infinite loop should handle interruption signals for clean shutdown in production environments.

+$shutdown = false;
+pcntl_signal(SIGTERM, function() use (&$shutdown) {
+    $shutdown = true;
+});
+pcntl_signal(SIGINT, function() use (&$shutdown) {
+    $shutdown = true;
+});
+
-while ($task = $consumer->waitTask()) {
+while (!$shutdown && ($task = $consumer->waitTask())) {
     try {
+        pcntl_signal_dispatch();
         // ... existing code ...
     }
 }

natsjobs/listener.go (1)

Line range hint 38-156: Consider refactoring for improved maintainability.

The listenerStart function is complex and handles multiple responsibilities. Consider breaking it down into smaller, focused functions:

1. Message metadata handling
2. Error state management
3. Item preparation
4. Acknowledgment handling

Example refactor:

func (c *Driver) listenerStart() {
    go func() {
        for {
            select {
            case m, ok := <-c.msgCh:
                if !ok {
                    c.handleClosedChannel()
                    return
                }
-               meta, err := m.Metadata()
-               if err != nil {
-                   errn := m.Nak()
-                   if errn != nil {
-                       c.log.Error("failed to send Nak state", zap.Error(errn), zap.Error(err))
-                       continue
-                   }
-                   c.log.Info("can't get message metadata", zap.Error(err))
-                   continue
-               }
+               meta, err := c.handleMessageMetadata(m)
+               if err != nil {
+                   continue
+               }

-               err = m.InProgress()
-               if err != nil {
-                   errn := m.Nak()
-                   // ... error handling
-               }
+               if err := c.setMessageInProgress(m); err != nil {
+                   continue
+               }

                item := &Item{}
                c.unpack(m.Data(), m.Headers(), item)
+               if err := c.prepareItem(item, m, meta); err != nil {
+                   continue
+               }

+               c.processItem(item, m)

            case <-c.stopCh:
                c.handleShutdown()
                return
            }
        }
    }()
}

func (c *Driver) handleMessageMetadata(m jetstream.Msg) (*jetstream.MsgMetadata, error) {
    // Extract metadata handling
}

func (c *Driver) setMessageInProgress(m jetstream.Msg) error {
    // Handle InProgress state
}

func (c *Driver) prepareItem(item *Item, m jetstream.Msg, meta *jetstream.MsgMetadata) error {
    // Prepare item with options and headers
}

func (c *Driver) processItem(item *Item, m jetstream.Msg) {
    // Handle item processing and acknowledgment
}

natsjobs/driver.go (1)

Line range hint 516-532: Critical: Headers are lost during message requeue

The requeue method is not preserving the message headers when republishing the message. This is inconsistent with the Push method's behavior and will lead to loss of header information during requeuing operations.

Apply this diff to maintain header consistency:

 _, err = c.jetstream.PublishMsg(context.Background(), &nats.Msg{
     Subject: c.subject,
     Data:    data,
+    Header:  item.headers,
 })

tests/jobs_nats_test.go (1)

122-127: Improve error handling in cleanup.

The cleanup function swallows errors and only logs them. Consider failing the test if cleanup fails, as this could affect subsequent tests.

 t.Cleanup(func() {
     errc := helpers.CleanupNats("nats://127.0.0.1:4222", "headers-test")
     if errc != nil {
-        t.Log(errc)
+        t.Errorf("cleanup failed: %v", errc)
     }
 })

Configuration used: CodeRabbit UI
Review profile: CHILL

Learnt from: KernelMrex
PR: roadrunner-server/nats#102
File: tests/jobs_nats_test.go:1027-1151
Timestamp: 2024-11-12T10:53:06.414Z
Learning: The user clarified that the header check for the NATS message subject is performed in a PHP test file, not in the Go test file. The Go test ensures that the NATS message is published and the pipeline processes the job correctly.

Learnt from: KernelMrex
PR: roadrunner-server/nats#102
File: tests/jobs_nats_test.go:1027-1151
Timestamp: 2024-11-12T10:53:06.414Z
Learning: The user clarified that the header check for the NATS message subject is performed in a PHP test file, not in the Go test file. The Go test ensures that the NATS message is published and the pipeline processes the job correctly.

Learnt from: KernelMrex
PR: roadrunner-server/nats#102
File: tests/jobs_nats_test.go:1027-1151
Timestamp: 2024-11-12T10:53:06.414Z
Learning: The user clarified that the header check for the NATS message subject is performed in a PHP test file, not in the Go test file. The Go test ensures that the NATS message is published and the pipeline processes the job correctly.

func (c *Driver) unpack(data []byte, headers map[string][]string, item *Item) {
+       // Create a defensive copy of headers
+       headersCopy := make(map[string][]string, len(headers))
+       if headers != nil {
+           for k, v := range headers {
+               vCopy := make([]string, len(v))
+               copy(vCopy, v)
+               headersCopy[k] = vCopy
+           }
+       }
        err := json.Unmarshal(data, item)
-       item.headers = headers
+       item.headers = headersCopy
        if err != nil {
            *item = Item{
                Job:     auto,
                Ident:   uuid.NewString(),
                Payload: data,
-               headers: headers,
+               headers: headersCopy,
                Options: &Options{

#!/bin/bash
# Description: Check if Consumer class typically requires configuration
# Look for other test files using Consumer to see if they pass config

# Search for Consumer initialization patterns in test files
rg -t php "new.*Consumer.*\(" tests/

#!/bin/bash
# Search for Consumer class implementation and constructor
ast-grep --pattern 'class Consumer {
  $$$
  public function __construct($$$) {
    $$$
  }
  $$$
}'

#!/bin/bash
# Search for Consumer class implementation and any header-related methods
rg -l "class Consumer" . | xargs rg -A 10 "class Consumer"

# Also search for header-related methods in Consumer
rg -i "header" . | grep -i "consumer"

#!/bin/bash
# Description: Verify the unpack method signature matches the new call.
# Expected: The unpack method should accept three parameters (data, headers, and item).

ast-grep --pattern 'func $_(c *Driver) unpack($_, $_, $_) {
  $$$
}'

#!/bin/bash
# Let's try a different approach to find the unpack method
# First, let's find all Go files that might contain the unpack method
rg -l "unpack" --type go

# Then, let's look for the actual implementation with context
rg "func.*unpack" --type go -A 5

#!/bin/bash
# Description: Check test coverage for header-related functionality

# Look for header-related tests
rg -l "TestNATS.*Header" tests/

# Look for header-related test helper functions
rg -l "WithHeaders?" tests/helpers/

# Check for header-related assertions
rg "assert.*Header" tests/

#!/bin/bash
# Let's examine the actual test implementation and assertions

# Check the content of the NATS header test
rg -A 10 "TestNATSHeaders" tests/

# Look for any header-related setup or verification in test files
rg -i "header" tests/ -A 5

# Check for any header-related functionality in the main code
rg -i "header" pkg/ -A 5

tests/php_test_files/jobs/jobs_ok.php (1)

19-21: Consider adding header validation to the test.

Since this PR focuses on proper header handling, consider enhancing this test to validate that headers are correctly passed through the system.

Example enhancement:

while ($task = $consumer->waitTask()) {
    try {
+       // Validate headers are properly passed
+       $headers = $task->getHeaders();
+       if (empty($headers)) {
+           throw new \RuntimeException('Expected headers to be present');
+       }
        $task->ack();
    } catch (\Throwable $e) {
        $task->error((string)$e);
    }
}

tests/php_test_files/jobs/jobs_ok_pq.php (1)

19-19: Document the purpose of the sleep duration

The hardcoded 15-second sleep appears to be for testing purposes, but its specific purpose isn't clear.

Consider adding a comment explaining why this specific duration was chosen:

+    // Simulate a long-running task with a 15-second delay
     sleep(15);

tests/php_test_files/jobs/jobs_ok_sleep1.php (1)

19-19: Document the purpose of sleep(1)

Consider adding a comment explaining that the sleep is intentionally added to simulate processing time in the test.

+    // Simulate task processing time
     sleep(1);

tests/php_test_files/jobs/jobs_ok_slow.php (1)

18-20: Consider reducing sleep duration for faster tests

The 60-second sleep in this test file might significantly slow down the test suite execution. Consider:

1. Reducing the sleep duration for regular testing
2. Using environment variables to control the sleep duration
3. Adding a comment explaining why this specific duration is necessary

while ($task = $consumer->waitTask()) {
    try {
-        sleep(60);
+        // Sleep duration can be configured via environment variable
+        $sleepDuration = (int)getenv('TEST_SLEEP_DURATION') ?: 5; // Default to 5 seconds
+        sleep($sleepDuration);

tests/php_test_files/jobs/jobs_ok_slow_rand.php (1)

Line range hint 19-23: Document the sleep condition's purpose.

The random sleep condition (>995) introduces a 60-second delay in ~0.5% of cases. This helps test timeout handling and slow task processing scenarios.

Consider adding a comment explaining this test behavior:

        $val = random_int(0, 1000);
+       // Simulate slow processing in ~0.5% of cases to test timeout handling
        if ($val > 995) {
            sleep(60);
        }

tests/php_test_files/jobs/jobs_ok_queue_name_exist.php (1)

Line range hint 1-24: Enhance test coverage for header handling

Given that this PR's main objective is to fix user-defined headers passing, this test file could be enhanced to verify header preservation during task processing.

Consider adding header validation to this test:

 while ($task = $consumer->waitTask()) {
     try {
         if ('unknown' === $task->getQueue()) {
             throw new RuntimeException('Queue name was not found');
         }
+        // Verify headers are preserved
+        $headers = $task->getHeaders();
+        if (empty($headers)) {
+            throw new RuntimeException('Expected headers not found');
+        }
 
         $task->ack();
     } catch (\Throwable $e) {
         $task->error((string)$e);
     }
 }

tests/php_test_files/jobs/jobs_err.php (1)

26-26: LGTM: Enhanced retry mechanism with proper header preservation

The updated implementation correctly:

• Preserves attempt count via headers
• Adds a 5-second delay between retries
• Uses nack() to indicate processing failure while requesting requeue

Consider documenting the retry delay (5 seconds) and max attempts (3) as configuration parameters rather than hard-coding them, making them adjustable for different environments or job types.

tests/php_test_files/jobs/jobs_create_memory.php (1)

Line range hint 1-47: LGTM with verification needed

The changes align with the PR's objective of fixing user-defined header handling. The test file provides good coverage by testing both memory and local queues with headers.

Consider adding explicit assertions to verify that headers are correctly preserved after task acknowledgment, as this is the main focus of the PR.

Configuration used: CodeRabbit UI
Review profile: CHILL

#!/bin/bash
# Description: Check for any remaining usage of the old `complete()` method
# and confirm consistent adoption of `ack()` across test files

echo "Checking for any remaining complete() calls:"
rg "\->complete\(\)" "tests/php_test_files"

echo -e "\nVerifying ack() usage in test files:"
rg "\->ack\(\)" "tests/php_test_files"

#!/bin/bash
# Description: Check for any remaining usage of complete() method in PHP test files
# Expected: No occurrences of ->complete() should be found, confirming consistent migration to ->ack()

rg -g "*.php" "->complete\(\)" tests/

#!/bin/bash
# Description: Check for any remaining usage of complete() method in PHP test files
# Escaping special characters in the pattern
rg -g "*.php" "\->complete\(\)" tests/

# Also check for ack() usage to confirm the migration
rg -g "*.php" "\->ack\(\)" tests/

#!/bin/bash
# Description: Check for any remaining usage of the old $task->complete() method
# and verify consistent usage of $task->ack()

echo "Checking for any remaining usage of ->complete()..."
rg -t php "->complete\(\)" tests/

echo "Verifying consistent usage of ->ack()..."
rg -t php "->ack\(\)" tests/

#!/bin/bash
# Description: Check for any remaining usage of the old $task->complete() method
# and verify consistent usage of $task->ack()

echo "Checking for any remaining usage of complete()..."
rg "\\\$task->complete\(\)" tests/

echo "Verifying consistent usage of ack()..."
rg "\\\$task->ack\(\)" tests/

#!/bin/bash
# Description: Verify that all test files use the new ack() method consistently
# and that there are no remaining instances of the old complete() method

# Search for any remaining complete() calls in PHP test files
echo "Checking for any remaining complete() calls:"
rg "->complete\(\)" "tests/php_test_files"

# Verify consistent usage of ack() in other test files
echo "Verifying ack() usage in test files:"
rg "->ack\(\)" "tests/php_test_files"

#!/bin/bash
# Description: Verify that all test files use the new ack() method consistently
# and that there are no remaining instances of the old complete() method

# Search for any remaining complete() calls in PHP test files
echo "Checking for any remaining complete() calls:"
rg "\->complete\(" "tests/php_test_files"

# Verify consistent usage of ack() in other test files
echo "Verifying ack() usage in test files:"
rg "\->ack\(" "tests/php_test_files"

#!/bin/bash
# Description: Check for any remaining instances of the old method name
# and verify consistent usage of the new method across PHP test files

# Check for any remaining $task->complete() calls
echo "Checking for remaining complete() calls:"
rg -l '\$task->complete\(\)' 'tests/php_test_files'

# Verify consistent usage of ack()
echo -e "\nVerifying ack() usage:"
rg -l '\$task->ack\(\)' 'tests/php_test_files'

#!/bin/bash
# Description: Check if any test files still use the old complete()/fail() methods
# Expected: No occurrences of the old methods

echo "Checking for any remaining uses of old methods..."
rg -l 'complete\(\)' 'tests/php_test_files'
rg -l 'fail\(' 'tests/php_test_files'

#!/bin/bash
# Let's try a more comprehensive search with context and without exact method call matching
echo "Searching for potential method names with context..."
rg -i "complete|fail|ack|nack" "tests/php_test_files" -B 2 -A 2

# Also search for task-related class definitions to understand the interface
echo -e "\nSearching for task-related class definitions..."
ast-grep --pattern 'class $_ {
  $$$
  complete($$$) {
    $$$
  }
  $$$
}'

ast-grep --pattern 'class $_ {
  $$$
  fail($$$) {
    $$$
  }
  $$$
}'

#!/bin/bash
# Description: Check for any remaining usage of complete() method and verify ack() implementation

# Check for any remaining complete() calls in PHP test files
echo "Checking for remaining complete() calls:"
rg "->complete\(\)" "tests/php_test_files"

# Check for consistent ack() implementation
echo "Verifying ack() implementation:"
rg "->ack\(\)" "tests/php_test_files"

# Look for the interface/class definition to verify the method change
echo "Looking for interface definition:"
rg "interface.*Task" -A 10

#!/bin/bash
# Let's modify our search patterns to escape special characters properly

# Check for complete() method usage
echo "Checking for complete() calls:"
rg "complete\(\)" "tests/php_test_files"

# Check for ack() method usage
echo "Checking for ack() calls:"
rg "ack\(\)" "tests/php_test_files"

# Look for Task interface or class definitions
echo "Looking for Task definitions:"
rg "interface.*Task|class.*Task" "tests/php_test_files"

# Check the specific file content to understand the context better
echo "Content of the test file:"
cat "tests/php_test_files/jobs/jobs_create_memory.php"

#!/bin/bash
# Description: Verify header handling implementation

# Look for header-related code in the main implementation
echo "Checking header handling implementation:"
rg "withHeader|getHeader" -A 2

# Check for NATS-specific header handling
echo "Checking NATS-specific header handling:"
ast-grep --pattern 'nats.NewMsg($$$).Header($$$)'

#!/bin/bash
# Look for task-related files and their header implementations
echo "Checking task-related files:"
fd "task" --type f

echo "\nChecking header implementation in task files:"
rg "class.*Task" -A 10

echo "\nChecking header-related method implementations:"
ast-grep --pattern 'function withHeader($$$) {
  $$$
}'

#!/bin/bash
# Search for header-related implementations in PHP files
echo "Checking PHP files for header implementations:"
rg -t php "class.*implements.*" -A 15

# Look for message handling in test files
echo "\nChecking test files for message handling:"
rg -t php "function.*handle.*message" -A 10

# Check for header usage in test assertions
echo "\nChecking test assertions with headers:"
rg -t php "assert.*header" -A 5