diff --git a/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts b/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts
new file mode 100644
index 000000000000..85214cbf17e2
--- /dev/null
+++ b/packages/beacon-node/src/chain/blocks/blockInput/blockInput.ts
@@ -0,0 +1,802 @@
+import {ForkName, ForkPreDeneb} from "@lodestar/params";
+import {BlobIndex, ColumnIndex, SignedBeaconBlock, Slot, deneb, fulu} from "@lodestar/types";
+import {fromHex, prettyBytes, toHex, withTimeout} from "@lodestar/utils";
+import {VersionedHashes} from "../../../execution/index.js";
+import {kzgCommitmentToVersionedHash} from "../../../util/blobs.js";
+import {byteArrayEquals} from "../../../util/bytes.js";
+import {BlockInputError, BlockInputErrorCode} from "./errors.js";
+import {
+  AddBlob,
+  AddBlock,
+  AddColumn,
+  BlobMeta,
+  BlobWithSource,
+  BlockInputInit,
+  ColumnMeta,
+  ColumnWithSource,
+  CreateBlockInputMeta,
+  DAData,
+  DAType,
+  IBlockInput,
+  LogMetaBasic,
+  LogMetaBlobs,
+  LogMetaColumns,
+  PromiseParts,
+  SourceMeta,
+} from "./types.js";
+
+export type BlockInput = BlockInputPreData | BlockInputBlobs | BlockInputColumns;
+
+export function createPromise<T>(): PromiseParts<T> {
+  let resolve!: (value: T) => void;
+  let reject!: (e: Error) => void;
+  const promise = new Promise<T>((_resolve, _reject) => {
+    resolve = _resolve;
+    reject = _reject;
+  });
+  return {
+    promise,
+    resolve,
+    reject,
+  };
+}
+
+type BlockInputState<F extends ForkName> =
+  | {
+      hasBlock: false;
+      hasAllData: false;
+    }
+  | {
+      hasBlock: false;
+      hasAllData: true;
+    }
+  | {
+      hasBlock: true;
+      hasAllData: false;
+      block: SignedBeaconBlock<F>;
+      source: SourceMeta;
+    }
+  | {
+      hasBlock: true;
+      hasAllData: true;
+      block: SignedBeaconBlock<F>;
+      source: SourceMeta;
+      timeCompleteSec: number;
+    };
+
+abstract class AbstractBlockInput<F extends ForkName = ForkName, TData extends DAData = DAData>
+  implements IBlockInput<F, TData>
+{
+  abstract type: DAType;
+  daOutOfRange: boolean;
+  timeCreated: number;
+
+  forkName: ForkName;
+  slot: Slot;
+  blockRootHex: string;
+  parentRootHex: string;
+
+  abstract state: BlockInputState<F>;
+
+  protected blockPromise = createPromise<SignedBeaconBlock<F>>();
+  protected dataPromise = createPromise<TData>();
+
+  constructor(init: BlockInputInit) {
+    this.daOutOfRange = init.daOutOfRange;
+    this.timeCreated = init.timeCreated;
+    this.forkName = init.forkName;
+    this.slot = init.slot;
+    this.blockRootHex = init.blockRootHex;
+    this.parentRootHex = init.parentRootHex;
+  }
+
+  abstract addBlock(props: AddBlock<F>): void;
+
+  hasBlock(): boolean {
+    return this.state.hasBlock;
+  }
+
+  getBlock(): SignedBeaconBlock<F> {
+    if (!this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISSING_BLOCK,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot getBlock from BlockInput without a block"
+      );
+    }
+    return this.state.block;
+  }
+
+  getBlockSource(): SourceMeta {
+    if (!this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISSING_BLOCK,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot getBlockSource from BlockInput without a block"
+      );
+    }
+    return this.state.source;
+  }
+
+  hasAllData(): boolean {
+    return this.state.hasAllData;
+  }
+
+  hasBlockAndAllData(): boolean {
+    return this.state.hasBlock && this.state.hasAllData;
+  }
+
+  getLogMeta(): LogMetaBasic {
+    return {
+      blockRoot: prettyBytes(this.blockRootHex),
+      slot: this.slot,
+    };
+  }
+
+  getTimeComplete(): number {
+    if (!this.state.hasBlock || !this.state.hasAllData) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISSING_TIME_COMPLETE,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot getTimeComplete from BlockInput without a block and data"
+      );
+    }
+    return this.state.timeCompleteSec;
+  }
+
+  waitForBlock(timeout: number, signal?: AbortSignal): Promise<SignedBeaconBlock<F>> {
+    if (!this.state.hasBlock) {
+      return withTimeout(() => this.blockPromise.promise, timeout, signal);
+    }
+    return Promise.resolve(this.state.block);
+  }
+  waitForAllData(timeout: number, signal?: AbortSignal): Promise<TData> {
+    return withTimeout(() => this.dataPromise.promise, timeout, signal);
+  }
+
+  async waitForBlockAndAllData(timeout: number, signal?: AbortSignal): Promise<this> {
+    if (!this.state.hasBlock || !this.state.hasAllData) {
+      await withTimeout(() => Promise.all([this.blockPromise.promise, this.dataPromise.promise]), timeout, signal);
+    }
+    return this;
+  }
+}
+
+// Pre-DA
+
+type BlockInputPreDataState = {
+  hasBlock: true;
+  hasAllData: true;
+  block: SignedBeaconBlock<ForkPreDeneb>;
+  source: SourceMeta;
+  timeCompleteSec: number;
+};
+
+/**
+ * Pre-DA, BlockInput only has a single state.
+ * - the block simply exists
+ */
+export class BlockInputPreData extends AbstractBlockInput<ForkPreDeneb, null> {
+  type = DAType.PreData as const;
+
+  state: BlockInputPreDataState;
+
+  private constructor(init: BlockInputInit, state: BlockInputPreDataState) {
+    super(init);
+    this.state = state;
+  }
+
+  static createFromBlock(props: AddBlock & CreateBlockInputMeta): BlockInputPreData {
+    const init: BlockInputInit = {
+      daOutOfRange: props.daOutOfRange,
+      timeCreated: props.source.seenTimestampSec,
+      forkName: props.forkName,
+      slot: props.block.message.slot,
+      blockRootHex: props.blockRootHex,
+      parentRootHex: toHex(props.block.message.parentRoot),
+    };
+    const state: BlockInputPreDataState = {
+      hasBlock: true,
+      hasAllData: true,
+      block: props.block,
+      source: props.source,
+      timeCompleteSec: props.source.seenTimestampSec,
+    };
+    return new BlockInputPreData(init, state);
+  }
+
+  addBlock(_: AddBlock): void {
+    throw new BlockInputError(
+      {
+        code: BlockInputErrorCode.INVALID_CONSTRUCTION,
+        blockRoot: this.blockRootHex,
+      },
+      "Cannot addBlock to BlockInputPreData"
+    );
+  }
+}
+
+// Blobs DA
+
+export type ForkBlobsDA = ForkName.deneb | ForkName.electra;
+
+type BlockInputBlobsState =
+  | {
+      hasBlock: true;
+      hasAllData: true;
+      versionedHashes: VersionedHashes;
+      block: SignedBeaconBlock<ForkBlobsDA>;
+      source: SourceMeta;
+      timeCompleteSec: number;
+    }
+  | {
+      hasBlock: true;
+      hasAllData: false;
+      versionedHashes: VersionedHashes;
+      block: SignedBeaconBlock<ForkBlobsDA>;
+      source: SourceMeta;
+    }
+  | {
+      hasBlock: false;
+      hasAllData: false;
+    };
+
+/**
+ * With blobs, BlockInput has several states:
+ * - The block is seen and all blobs are seen
+ * - The block is seen and all blobs are not yet seen
+ * - The block is yet not seen and its unknown if all blobs are seen
+ */
+export class BlockInputBlobs extends AbstractBlockInput<ForkBlobsDA, deneb.BlobSidecars> {
+  type = DAType.Blobs as const;
+
+  state: BlockInputBlobsState;
+  private blobsCache = new Map<BlobIndex, BlobWithSource>();
+
+  private constructor(init: BlockInputInit, state: BlockInputBlobsState) {
+    super(init);
+    this.state = state;
+  }
+
+  static createFromBlock(props: AddBlock<ForkBlobsDA> & CreateBlockInputMeta): BlockInputBlobs {
+    const hasAllData = props.daOutOfRange || props.block.message.body.blobKzgCommitments.length === 0;
+
+    const state = {
+      hasBlock: true,
+      hasAllData,
+      versionedHashes: props.block.message.body.blobKzgCommitments.map(kzgCommitmentToVersionedHash),
+      block: props.block,
+      source: props.source,
+      timeCompleteSec: hasAllData ? props.source.seenTimestampSec : undefined,
+    } as BlockInputBlobsState;
+    const init: BlockInputInit = {
+      daOutOfRange: props.daOutOfRange,
+      timeCreated: props.source.seenTimestampSec,
+      forkName: props.forkName,
+      slot: props.block.message.slot,
+      blockRootHex: props.blockRootHex,
+      parentRootHex: toHex(props.block.message.parentRoot),
+    };
+    const blockInput = new BlockInputBlobs(init, state);
+    blockInput.blockPromise.resolve(props.block);
+    if (hasAllData) {
+      blockInput.dataPromise.resolve([]);
+    }
+    return blockInput;
+  }
+
+  static createFromBlob(props: AddBlob & CreateBlockInputMeta): BlockInputBlobs {
+    const state: BlockInputBlobsState = {
+      hasBlock: false,
+      hasAllData: false,
+    };
+    const init: BlockInputInit = {
+      daOutOfRange: props.daOutOfRange,
+      timeCreated: props.seenTimestampSec,
+      forkName: props.forkName,
+      blockRootHex: props.blockRootHex,
+      parentRootHex: toHex(props.blobSidecar.signedBlockHeader.message.parentRoot),
+      slot: props.blobSidecar.signedBlockHeader.message.slot,
+    };
+    const blockInput = new BlockInputBlobs(init, state);
+    blockInput.blobsCache.set(props.blobSidecar.index, {
+      blobSidecar: props.blobSidecar,
+      source: props.source,
+      seenTimestampSec: props.seenTimestampSec,
+      peerIdStr: props.peerIdStr,
+    });
+    return blockInput;
+  }
+
+  getLogMeta(): LogMetaBlobs {
+    return {
+      blockRoot: prettyBytes(this.blockRootHex),
+      slot: this.slot,
+      expectedBlobs: this.state.hasBlock ? this.state.block.message.body.blobKzgCommitments.length : "unknown",
+      receivedBlobs: this.blobsCache.size,
+    };
+  }
+
+  addBlock({blockRootHex, block, source}: AddBlock<ForkBlobsDA>): void {
+    if (!this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INVALID_CONSTRUCTION,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot addBlock to BlockInputBlobs after it already has a block"
+      );
+    }
+
+    // this check suffices for checking slot, parentRoot, and forkName
+    if (blockRootHex !== this.blockRootHex) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISMATCHED_ROOT_HEX,
+          blockInputRoot: this.blockRootHex,
+          mismatchedRoot: blockRootHex,
+          source: source.source,
+          peerId: `${source.peerIdStr}`,
+        },
+        "addBlock blockRootHex does not match BlockInput.blockRootHex"
+      );
+    }
+
+    for (const {blobSidecar} of this.blobsCache.values()) {
+      if (!blockAndBlobArePaired(block, blobSidecar)) {
+        this.blobsCache.delete(blobSidecar.index);
+        // TODO: (@matthewkeil) spec says to ignore invalid blobs but should we downscore the peer maybe?
+        // this.logger?.error(`Removing blobIndex=${blobSidecar.index} from BlockInput`, {}, err);
+      }
+    }
+
+    const hasAllData = this.blobsCache.size === block.message.body.blobKzgCommitments.length;
+
+    this.state = {
+      ...this.state,
+      hasAllData,
+      block,
+      versionedHashes: block.message.body.blobKzgCommitments.map(kzgCommitmentToVersionedHash),
+      source,
+      timeCompleteSec: hasAllData ? source.seenTimestampSec : undefined,
+    } as BlockInputBlobsState;
+    this.blockPromise.resolve(block);
+    if (hasAllData) {
+      this.dataPromise.resolve(this.getBlobs());
+    }
+  }
+
+  hasBlob(blobIndex: BlobIndex): boolean {
+    return this.blobsCache.has(blobIndex);
+  }
+
+  addBlob({blockRootHex, blobSidecar, source, peerIdStr, seenTimestampSec}: AddBlob): void {
+    if (this.state.hasAllData) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INVALID_CONSTRUCTION,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot addBlob to BlockInputBlobs after it already is complete"
+      );
+    }
+    if (this.blobsCache.has(blobSidecar.index)) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INVALID_CONSTRUCTION,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot addBlob to BlockInputBlobs with duplicate blobIndex"
+      );
+    }
+
+    // this check suffices for checking slot, parentRoot, and forkName
+    if (blockRootHex !== this.blockRootHex) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISMATCHED_ROOT_HEX,
+          blockInputRoot: this.blockRootHex,
+          mismatchedRoot: blockRootHex,
+          source: source,
+          peerId: `${peerIdStr}`,
+        },
+        "Blob BeaconBlockHeader blockRootHex does not match BlockInput.blockRootHex"
+      );
+    }
+
+    if (this.state.hasBlock) {
+      assertBlockAndBlobArePaired(this.blockRootHex, this.state.block, blobSidecar);
+    }
+
+    this.blobsCache.set(blobSidecar.index, {blobSidecar, source, seenTimestampSec, peerIdStr});
+
+    if (this.state.hasBlock && this.blobsCache.size === this.state.block.message.body.blobKzgCommitments.length) {
+      this.state = {
+        ...this.state,
+        hasAllData: true,
+        timeCompleteSec: seenTimestampSec,
+      };
+      this.dataPromise.resolve([...this.blobsCache.values()].map(({blobSidecar}) => blobSidecar));
+    }
+  }
+
+  getVersionedHashes(): VersionedHashes {
+    if (!this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INCOMPLETE_DATA,
+          ...this.getLogMeta(),
+        },
+        "Cannot get versioned hashes. Block is unknown"
+      );
+    }
+    return this.state.versionedHashes;
+  }
+
+  getMissingBlobMeta(): BlobMeta[] {
+    if (!this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INCOMPLETE_DATA,
+          ...this.getLogMeta(),
+        },
+        "Cannot get missing blobs. Block is unknown"
+      );
+    }
+    if (this.state.hasAllData) {
+      return [];
+    }
+
+    const blobMeta: BlobMeta[] = [];
+    const versionedHashes = this.state.versionedHashes;
+    for (let index = 0; index < versionedHashes.length; index++) {
+      if (!this.blobsCache.has(index)) {
+        blobMeta.push({
+          index,
+          blockRoot: fromHex(this.blockRootHex),
+          versionHash: versionedHashes[index],
+        });
+      }
+    }
+    return blobMeta;
+  }
+
+  getAllBlobsWithSource(): BlobWithSource[] {
+    if (!this.state.hasAllData) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INCOMPLETE_DATA,
+          ...this.getLogMeta(),
+        },
+        "Cannot get all blobs. DA status is not complete"
+      );
+    }
+    return [...this.blobsCache.values()];
+  }
+
+  getBlobs(): deneb.BlobSidecars {
+    return this.getAllBlobsWithSource().map(({blobSidecar}) => blobSidecar);
+  }
+}
+
+function blockAndBlobArePaired(block: SignedBeaconBlock<ForkBlobsDA>, blobSidecar: deneb.BlobSidecar): boolean {
+  return byteArrayEquals(block.message.body.blobKzgCommitments[blobSidecar.index], blobSidecar.kzgCommitment);
+}
+
+function assertBlockAndBlobArePaired(
+  blockRootHex: string,
+  block: SignedBeaconBlock<ForkBlobsDA>,
+  blobSidecar: deneb.BlobSidecar
+): void {
+  if (!blockAndBlobArePaired(block, blobSidecar)) {
+    // TODO: (@matthewkeil) should this eject the bad blob instead? No way to tell if the blob or the block
+    //       has the invalid commitment. Guessing it would be the blob though because we match via block
+    //       hashTreeRoot and we do not take a hashTreeRoot of the BlobSidecar
+    throw new BlockInputError(
+      {
+        code: BlockInputErrorCode.MISMATCHED_KZG_COMMITMENT,
+        blockRoot: blockRootHex,
+        slot: block.message.slot,
+        sidecarIndex: blobSidecar.index,
+      },
+      "BlobSidecar commitment does not match block commitment"
+    );
+  }
+}
+
+// Columns DA
+
+export type ForkColumnsDA = ForkName.fulu;
+
+type BlockInputColumnsState =
+  | {
+      hasBlock: true;
+      hasAllData: true;
+      versionedHashes: VersionedHashes;
+      block: SignedBeaconBlock<ForkColumnsDA>;
+      source: SourceMeta;
+      timeCompleteSec: number;
+    }
+  | {
+      hasBlock: true;
+      hasAllData: false;
+      versionedHashes: VersionedHashes;
+      block: SignedBeaconBlock<ForkColumnsDA>;
+      source: SourceMeta;
+    }
+  | {
+      hasBlock: false;
+      hasAllData: true;
+      versionedHashes: VersionedHashes;
+    }
+  | {
+      hasBlock: false;
+      hasAllData: false;
+      versionedHashes: VersionedHashes;
+    };
+/**
+ * With columns, BlockInput has several states:
+ * - The block is seen and all required sampled columns are seen
+ * - The block is seen and all required sampled columns are not yet seen
+ * - The block is not yet seen and all required sampled columns are seen
+ * - The block is not yet seen and all required sampled columns are not yet seen
+ */
+export class BlockInputColumns extends AbstractBlockInput<ForkColumnsDA, fulu.DataColumnSidecars> {
+  type = DAType.Columns as const;
+
+  state: BlockInputColumnsState;
+
+  private columnsCache = new Map<ColumnIndex, ColumnWithSource>();
+  private readonly sampledColumns: ColumnIndex[];
+  private readonly custodyColumns: ColumnIndex[];
+
+  private constructor(
+    init: BlockInputInit,
+    state: BlockInputColumnsState,
+    sampledColumns: ColumnIndex[],
+    custodyColumns: ColumnIndex[]
+  ) {
+    super(init);
+    this.state = state;
+    this.sampledColumns = sampledColumns;
+    this.custodyColumns = custodyColumns;
+  }
+
+  static createFromBlock(
+    props: AddBlock<ForkColumnsDA> &
+      CreateBlockInputMeta & {sampledColumns: ColumnIndex[]; custodyColumns: ColumnIndex[]}
+  ): BlockInputColumns {
+    const hasAllData =
+      props.daOutOfRange ||
+      props.block.message.body.blobKzgCommitments.length === 0 ||
+      props.sampledColumns.length === 0;
+    const state = {
+      hasBlock: true,
+      hasAllData,
+      versionedHashes: props.block.message.body.blobKzgCommitments.map(kzgCommitmentToVersionedHash),
+      block: props.block,
+      source: props.source,
+      timeCreated: props.source.seenTimestampSec,
+      timeCompleteSec: hasAllData ? props.source.seenTimestampSec : undefined,
+    } as BlockInputColumnsState;
+    const init: BlockInputInit = {
+      daOutOfRange: props.daOutOfRange,
+      timeCreated: props.source.seenTimestampSec,
+      forkName: props.forkName,
+      blockRootHex: props.blockRootHex,
+      parentRootHex: toHex(props.block.message.parentRoot),
+      slot: props.block.message.slot,
+    };
+    const blockInput = new BlockInputColumns(init, state, props.sampledColumns, props.custodyColumns);
+
+    blockInput.blockPromise.resolve(props.block);
+    if (hasAllData) {
+      blockInput.dataPromise.resolve([]);
+    }
+    return blockInput;
+  }
+
+  static createFromColumn(
+    props: AddColumn & CreateBlockInputMeta & {sampledColumns: ColumnIndex[]; custodyColumns: ColumnIndex[]}
+  ): BlockInputColumns {
+    const hasAllData = props.sampledColumns.length === 0;
+    const state: BlockInputColumnsState = {
+      hasBlock: false,
+      hasAllData,
+      versionedHashes: props.columnSidecar.kzgCommitments.map(kzgCommitmentToVersionedHash),
+    };
+    const init: BlockInputInit = {
+      daOutOfRange: false,
+      timeCreated: props.seenTimestampSec,
+      forkName: props.forkName,
+      blockRootHex: props.blockRootHex,
+      parentRootHex: toHex(props.columnSidecar.signedBlockHeader.message.parentRoot),
+      slot: props.columnSidecar.signedBlockHeader.message.slot,
+    };
+    const blockInput = new BlockInputColumns(init, state, props.sampledColumns, props.custodyColumns);
+    if (hasAllData) {
+      blockInput.dataPromise.resolve([]);
+    }
+    return blockInput;
+  }
+
+  getLogMeta(): LogMetaColumns {
+    return {
+      blockRoot: prettyBytes(this.blockRootHex),
+      slot: this.slot,
+      expectedColumns:
+        this.state.hasBlock && this.state.block.message.body.blobKzgCommitments.length === 0
+          ? 0
+          : this.sampledColumns.length,
+      receivedColumns: this.getSampledColumns().length,
+    };
+  }
+
+  addBlock(props: AddBlock<ForkColumnsDA>): void {
+    if (this.state.hasBlock) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.INVALID_CONSTRUCTION,
+          blockRoot: this.blockRootHex,
+        },
+        "Cannot addBlock to BlockInputColumns after it already has a block"
+      );
+    }
+
+    if (props.blockRootHex !== this.blockRootHex) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISMATCHED_ROOT_HEX,
+          blockInputRoot: this.blockRootHex,
+          mismatchedRoot: props.blockRootHex,
+          source: props.source.source,
+          peerId: `${props.source.peerIdStr}`,
+        },
+        "addBlock blockRootHex does not match BlockInput.blockRootHex"
+      );
+    }
+
+    for (const {columnSidecar} of this.columnsCache.values()) {
+      if (!blockAndColumnArePaired(props.block, columnSidecar)) {
+        this.columnsCache.delete(columnSidecar.index);
+        // this.logger?.error(`Removing columnIndex=${columnSidecar.index} from BlockInput`, {}, err);
+      }
+    }
+
+    const hasAllData = props.block.message.body.blobKzgCommitments.length === 0 || this.state.hasAllData;
+
+    this.state = {
+      ...this.state,
+      hasBlock: true,
+      hasAllData,
+      block: props.block,
+      source: props.source,
+      timeCompleteSec: hasAllData ? props.source.seenTimestampSec : undefined,
+    } as BlockInputColumnsState;
+
+    this.blockPromise.resolve(props.block);
+  }
+
+  addColumn({blockRootHex, columnSidecar, source, seenTimestampSec, peerIdStr}: AddColumn): void {
+    if (blockRootHex !== this.blockRootHex) {
+      throw new BlockInputError(
+        {
+          code: BlockInputErrorCode.MISMATCHED_ROOT_HEX,
+          blockInputRoot: this.blockRootHex,
+          mismatchedRoot: blockRootHex,
+          source: source,
+          peerId: `${peerIdStr}`,
+        },
+        "Column BeaconBlockHeader blockRootHex does not match BlockInput.blockRootHex"
+      );
+    }
+
+    if (this.state.hasBlock) {
+      assertBlockAndColumnArePaired(this.blockRootHex, this.state.block, columnSidecar);
+    }
+
+    this.columnsCache.set(columnSidecar.index, {columnSidecar, source, seenTimestampSec, peerIdStr});
+
+    const sampledColumns = this.getSampledColumns();
+    const hasAllData = this.state.hasAllData || sampledColumns.length === this.sampledColumns.length;
+
+    this.state = {
+      ...this.state,
+      hasAllData: hasAllData || this.state.hasAllData,
+      timeCompleteSec: hasAllData ? seenTimestampSec : undefined,
+    } as BlockInputColumnsState;
+
+    if (hasAllData && sampledColumns !== null) {
+      this.dataPromise.resolve(sampledColumns);
+    }
+  }
+
+  hasColumn(columnIndex: number): boolean {
+    return this.columnsCache.has(columnIndex);
+  }
+
+  getVersionedHashes(): VersionedHashes {
+    return this.state.versionedHashes;
+  }
+
+  getCustodyColumns(): fulu.DataColumnSidecars {
+    const columns: fulu.DataColumnSidecars = [];
+    for (const index of this.custodyColumns) {
+      const column = this.columnsCache.get(index);
+      if (column) {
+        columns.push(column.columnSidecar);
+      }
+    }
+    return columns;
+  }
+
+  getSampledColumns(): fulu.DataColumnSidecars {
+    const columns: fulu.DataColumnSidecars = [];
+    for (const index of this.sampledColumns) {
+      const column = this.columnsCache.get(index);
+      if (column) {
+        columns.push(column.columnSidecar);
+      }
+    }
+    return columns;
+  }
+
+  getAllColumnsWithSource(): ColumnWithSource[] {
+    return [...this.columnsCache.values()];
+  }
+
+  getAllColumns(): fulu.DataColumnSidecars {
+    return this.getAllColumnsWithSource().map(({columnSidecar}) => columnSidecar);
+  }
+
+  getMissingSampledColumnMeta(): ColumnMeta[] {
+    if (this.state.hasAllData) {
+      return [];
+    }
+
+    const needed: ColumnMeta[] = [];
+    const blockRoot = fromHex(this.blockRootHex);
+    for (const index of this.sampledColumns) {
+      if (!this.columnsCache.has(index)) {
+        needed.push({index, blockRoot});
+      }
+    }
+    return needed;
+  }
+}
+
+function blockAndColumnArePaired(
+  block: SignedBeaconBlock<ForkColumnsDA>,
+  columnSidecar: fulu.DataColumnSidecar
+): boolean {
+  return (
+    block.message.body.blobKzgCommitments.length === columnSidecar.kzgCommitments.length &&
+    block.message.body.blobKzgCommitments.every((commitment, index) =>
+      byteArrayEquals(commitment, columnSidecar.kzgCommitments[index])
+    )
+  );
+}
+
+function assertBlockAndColumnArePaired(
+  blockRootHex: string,
+  block: SignedBeaconBlock<ForkColumnsDA>,
+  columnSidecar: fulu.DataColumnSidecar
+): void {
+  if (!blockAndColumnArePaired(block, columnSidecar)) {
+    throw new BlockInputError(
+      {
+        code: BlockInputErrorCode.MISMATCHED_KZG_COMMITMENT,
+        blockRoot: blockRootHex,
+        slot: block.message.slot,
+        sidecarIndex: columnSidecar.index,
+      },
+      "DataColumnsSidecar kzgCommitment does not match block kzgCommitment"
+    );
+  }
+}
diff --git a/packages/beacon-node/src/chain/blocks/blockInput/errors.ts b/packages/beacon-node/src/chain/blocks/blockInput/errors.ts
new file mode 100644
index 000000000000..b30f16a66e22
--- /dev/null
+++ b/packages/beacon-node/src/chain/blocks/blockInput/errors.ts
@@ -0,0 +1,48 @@
+import {Slot} from "@lodestar/types";
+import {LodestarError} from "@lodestar/utils";
+import {PeerIdStr} from "../../../util/peerId.js";
+import {BlockInputSource, LogMetaBlobs, LogMetaColumns} from "./types.js";
+
+export enum BlockInputErrorCode {
+  // Bad Arguments
+  INVALID_CONSTRUCTION = "BLOCK_INPUT_ERROR_INVALID_CONSTRUCTION",
+
+  // Attempt to get all data but some is missing
+  INCOMPLETE_DATA = "BLOCK_INPUT_ERROR_INCOMPLETE_DATA",
+
+  // Missing class property values for getters
+  MISSING_BLOCK = "BLOCK_INPUT_ERROR_MISSING_BLOCK",
+  MISSING_TIME_COMPLETE = "BLOCK_INPUT_ERROR_MISSING_TIME_COMPLETE",
+
+  // Mismatched values
+  MISMATCHED_ROOT_HEX = "BLOCK_INPUT_ERROR_MISMATCHED_ROOT_HEX",
+  MISMATCHED_KZG_COMMITMENT = "BLOCK_INPUT_ERROR_MISMATCHED_KZG_COMMITMENT",
+}
+
+export type BlockInputErrorType =
+  | {
+      code: BlockInputErrorCode.MISSING_BLOCK | BlockInputErrorCode.MISSING_TIME_COMPLETE;
+      blockRoot: string;
+    }
+  | {
+      code: BlockInputErrorCode.INVALID_CONSTRUCTION;
+      blockRoot: string;
+    }
+  | {
+      code: BlockInputErrorCode.MISMATCHED_ROOT_HEX;
+      blockInputRoot: string;
+      mismatchedRoot: string;
+      source: BlockInputSource;
+      peerId: PeerIdStr;
+    }
+  | {
+      code: BlockInputErrorCode.MISMATCHED_KZG_COMMITMENT;
+      blockRoot: string;
+      slot: undefined | Slot;
+      sidecarIndex: number;
+      commitmentIndex?: number;
+    }
+  | (LogMetaBlobs & {code: BlockInputErrorCode.INCOMPLETE_DATA})
+  | (LogMetaColumns & {code: BlockInputErrorCode.INCOMPLETE_DATA});
+
+export class BlockInputError extends LodestarError<BlockInputErrorType> {}
diff --git a/packages/beacon-node/src/chain/blocks/blockInput/index.ts b/packages/beacon-node/src/chain/blocks/blockInput/index.ts
new file mode 100644
index 000000000000..f3cea1e241eb
--- /dev/null
+++ b/packages/beacon-node/src/chain/blocks/blockInput/index.ts
@@ -0,0 +1,3 @@
+export * from "./blockInput.js";
+export * from "./errors.js";
+export * from "./types.js";
diff --git a/packages/beacon-node/src/chain/blocks/blockInput/types.ts b/packages/beacon-node/src/chain/blocks/blockInput/types.ts
new file mode 100644
index 000000000000..37dc11bafc11
--- /dev/null
+++ b/packages/beacon-node/src/chain/blocks/blockInput/types.ts
@@ -0,0 +1,137 @@
+import {ForkName} from "@lodestar/params";
+import {RootHex, SignedBeaconBlock, Slot, deneb, fulu} from "@lodestar/types";
+
+export enum DAType {
+  PreData = "pre-data",
+  Blobs = "blobs",
+  Columns = "columns",
+}
+
+export type DAData = null | deneb.BlobSidecars | fulu.DataColumnSidecars;
+
+/**
+ * Represents were input originated. Blocks and Data can come from different
+ * sources so each should be labelled individually.
+ */
+export enum BlockInputSource {
+  gossip = "gossip",
+  api = "api",
+  engine = "engine",
+  byRange = "req_resp_by_range",
+  byRoot = "req_resp_by_root",
+}
+
+export type PromiseParts<T> = {
+  promise: Promise<T>;
+  resolve: (value: T) => void;
+  reject: (e: Error) => void;
+};
+
+export type LogMetaBasic = {
+  slot: number;
+  blockRoot: string;
+};
+
+export type LogMetaBlobs = LogMetaBasic & {
+  expectedBlobs: number | string;
+  receivedBlobs: number;
+};
+
+export type LogMetaColumns = LogMetaBasic & {
+  expectedColumns: number;
+  receivedColumns: number;
+};
+
+export type SourceMeta = {
+  source: BlockInputSource;
+  seenTimestampSec: number;
+  peerIdStr?: string;
+};
+
+export type BlobWithSource = SourceMeta & {blobSidecar: deneb.BlobSidecar};
+
+export type ColumnWithSource = SourceMeta & {columnSidecar: fulu.DataColumnSidecar};
+
+export type BlockHeaderMeta = {
+  forkName: ForkName;
+  slot: Slot;
+  blockRootHex: string;
+  parentRootHex: string;
+};
+
+export type CreateBlockInputMeta = {
+  daOutOfRange: boolean;
+  forkName: ForkName;
+  blockRootHex: string;
+};
+
+export type BlockInputInit = BlockHeaderMeta & {
+  daOutOfRange: boolean;
+  timeCreated: number;
+};
+
+export type AddBlock<F extends ForkName = ForkName> = {
+  block: SignedBeaconBlock<F>;
+  blockRootHex: string;
+  source: SourceMeta;
+};
+
+export type AddBlob = BlobWithSource & {
+  blockRootHex: RootHex;
+};
+
+export type AddColumn = ColumnWithSource & {
+  blockRootHex: RootHex;
+};
+
+export type BlobMeta = ColumnMeta & {versionHash: Uint8Array};
+
+export type ColumnMeta = {
+  blockRoot: Uint8Array;
+  index: number;
+};
+
+/**
+ * This is used to validate that BlockInput implementations follow some minimal subset of operations
+ * and that adding a new implementation won't break consumers that rely on this subset.
+ *
+ * Practically speaking, this interface is only used internally.
+ */
+export interface IBlockInput<F extends ForkName = ForkName, TData extends DAData = DAData> {
+  type: DAType;
+
+  /** validator activities can't be performed on out of range data */
+  daOutOfRange: boolean;
+
+  timeCreated: number;
+  // block header metadata
+  forkName: ForkName;
+  slot: Slot;
+  blockRootHex: string;
+  parentRootHex: string;
+
+  addBlock(props: AddBlock<F>): void;
+  /** Whether the block has been seen and validated. If true, `getBlock` is guaranteed to not throw */
+  hasBlock(): boolean;
+  getBlock(): SignedBeaconBlock<F>;
+  getBlockSource(): SourceMeta;
+
+  /** Whether all expected DA data has been seen and validated. */
+  hasAllData(): boolean;
+
+  /**
+   * Whether the block and all DA data retrieved.
+   * If true, `getBlock` is guaranteed to not throw,
+   * and `getDAStatus` is guaranteed to be DAStatus.Complete
+   */
+  hasBlockAndAllData(): boolean;
+
+  getLogMeta(): LogMetaBasic;
+
+  /** Only safe to call when `hasBlockAndAllData` is true */
+  getTimeComplete(): number;
+
+  waitForBlock(timeout: number, signal?: AbortSignal): Promise<SignedBeaconBlock<F>>;
+  waitForAllData(timeout: number, signal?: AbortSignal): Promise<TData>;
+  waitForBlockAndAllData(timeout: number, signal?: AbortSignal): Promise<this>;
+}