feat(blobs): Limit blob txs to ones with full network representation (#…

…39)

## 📝 Summary

This is an alternative to the `no-fwd-blobs` branch that filters blobs
depending on whether or not they have the `BlobTxSidecar` available. If
the sidecar is available then the blob is in the correct network
representation form and can be collected and forwarded, otherwise the
blob data is incomplete and shouldn't be forwarded.

```
2024-06-21T09:17:06.111-0700	DEBUG	processTx	{"tx_hash": "0x82161fd324f41f29fa8fdf0720938b946b37ea7f6fa051f74305dbd384ed624d", "source": "alchemy"}
2024-06-21T09:17:06.112-0700	DEBUG	error: invalid blob transaction	{"tx_hash": "0x82161fd324f41f29fa8fdf0720938b946b37ea7f6fa051f74305dbd384ed624d", "source": "alchemy", "reason": "missing blob sidecar"}
```

Notably the "full" `newPendingTransactions` subscription from Alchemy
and Quiknode returns the incomplete blob. This is probably an upstream
`geth` bug I'll see if it's possible for the subscription to contain the
sidecar.

I haven't been able to test other node sources, but hopefully they
return the full network representation.

## ⛱ Motivation and Context

Blobs come in two forms, there's the `PoolTransactions` network
representation used in the mempool, and the blobless
`TransactionPayload` format used in block bodies.

## 📚 References

- https://eips.ethereum.org/EIPS/eip-4844#networking
- INFURA/go-ethlibs#92

---

## ✅ I have run these commands

* [x] `make lint`
* [x] `make test`
* [x] `go mod tidy`

collector/tx_processor.go

@@ -2,6 +2,7 @@ package collector
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -24,6 +25,8 @@ const (
 	receiverTimeout = 5 * time.Second
 )
 
+var ErrBlobMissingSidecar = errors.New("missing blob sidecar")
+
 type TxProcessorOpts struct {
 	Log                     *zap.SugaredLogger
 	OutDir                  string
@@ -285,6 +288,13 @@ func (p *TxProcessor) validateTx(fTrash *os.File, txIn TxIn) error { // inspired
 		return core.ErrTipAboveFeeCap
 	}
 
+	// Ensure blob txs are correctly formed
+	if err := p.validateBlobTx(tx); err != nil {
+		log.Debugw("error: invalid blob transaction", "reason", err)
+		p.writeTrash(fTrash, txIn, "invalid blob transaction", "")
+		return err
+	}
+
 	// all good
 	return nil
 }
@@ -440,3 +450,18 @@ func (p *TxProcessor) healthCheckCall() {
 	}
 	resp.Body.Close()
 }
+
+// validateBlobTx ensures that a blob tx is capable of being consumed
+// by our system.  Namely, the blob tx should be in the "full" PooledTransactions
+// network representation with the full sidecar available.
+func (p *TxProcessor) validateBlobTx(tx *types.Transaction) error {
+	if tx.Type() != types.BlobTxType {
+		return nil
+	}
+
+	if tx.BlobTxSidecar() == nil {
+		return ErrBlobMissingSidecar
+	}
+
+	return nil
+}